From f41595fb101feb428e3dc0bbaf8cfe5e49be67fc Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 17 Aug 2026 21:22:03 -0700 Subject: [PATCH] mount: drop consumed entries when reading a directory through (#10802) The cached readdir trims the head of the handle's entry stream as the client walks past it; the read-through path never did, so a directory too large to cache -- the only kind that takes that path -- was held whole in the handle for the length of the walk. Hoist the trim to cover both paths. --- weed/mount/weedfs_dir_read.go | 38 +++++--- weed/mount/weedfs_dir_read_pagination_test.go | 92 +++++++++++++++++++ .../weedfs_invalidate_open_handle_test.go | 2 +- 3 files changed, 118 insertions(+), 14 deletions(-) diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index 556626507..e5493001b 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -53,6 +53,26 @@ func (dh *DirectoryHandle) reset() { dh.entryStreamOffset = directoryStreamBaseOffset } +// dropConsumed releases the entries the client has already 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. +func (dh *DirectoryHandle) dropConsumed(offset uint64) { + if offset < dh.entryStreamOffset { + return + } + trim := int(offset-dh.entryStreamOffset) - 1 + if trim <= 0 || trim > len(dh.entryStream) { + return + } + 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) +} + type DirectoryHandleToInode struct { sync.Mutex dir2inode map[DirectoryHandleId]*DirectoryHandle @@ -251,25 +271,17 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode var lastEntryName string + // Both the cached and the direct walk page through the same stream, and a + // directory read through is precisely one too big to hold, so drop the + // consumed head before either of them appends to it. + dh.dropConsumed(input.Offset) + if wfs.inodeToPath.ShouldReadDirectoryDirect(dirPath) { return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) } // 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 { diff --git a/weed/mount/weedfs_dir_read_pagination_test.go b/weed/mount/weedfs_dir_read_pagination_test.go index 00e07d7e1..c2690b403 100644 --- a/weed/mount/weedfs_dir_read_pagination_test.go +++ b/weed/mount/weedfs_dir_read_pagination_test.go @@ -12,6 +12,7 @@ import ( "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/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -244,3 +245,94 @@ func TestReadDirTrimsConsumedEntries(t *testing.T) { t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total) } } + +// listingFilerServer serves a fixed, sorted child list the way the filer +// paginates one. +type listingFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + names []string +} + +func (s *listingFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { + var sent uint32 + for _, name := range s.names { + if name < req.StartFromFileName || (name == req.StartFromFileName && !req.InclusiveStartFrom) { + continue + } + if req.Limit > 0 && sent >= req.Limit { + break + } + if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: &filer_pb.Entry{ + Name: name, + Attributes: &filer_pb.FuseAttributes{FileMode: 0o644, FileSize: 1}, + }}); err != nil { + return err + } + sent++ + } + return nil +} + +// TestReadDirDirectTrimsConsumedEntries is the same check for a directory read +// through to the filer. That path is taken precisely for directories too big to +// cache, so holding the whole walk in the handle is where a listing turns into +// gigabytes of resident memory. +func TestReadDirDirectTrimsConsumedEntries(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, nil, 0) + startFakeFiler(t, wfs, &listingFilerServer{names: names}) + dirInode, _ := wfs.inodeToPath.GetInode(dir) + if !wfs.inodeToPath.MarkDirectoryReadThrough(dir, time.Now()) { + t.Fatal("directory did not enter read-through mode") + } + + dhid, dh := wfs.AcquireDirectoryHandle() + defer wfs.ReleaseDirectoryHandle(dhid) + + 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) + } + } + if peak >= total { + t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total) + } +} diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 260580259..b79e6d596 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -156,7 +156,7 @@ func (s *fakeFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateE } // startFakeFiler serves fake on a local port and points wfs at it. -func startFakeFiler(t *testing.T, wfs *WFS, fake *fakeFilerServer) { +func startFakeFiler(t *testing.T, wfs *WFS, fake filer_pb.SeaweedFilerServer) { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil {