From df879e1ed70f15bc7470f5f7edb0df9c159a76f2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 3 Jun 2026 10:28:42 -0700 Subject: [PATCH] filer: bound TraverseBfsMetadata memory by queuing directory paths (#9814) * filer: bound TraverseBfsMetadata memory by queuing directory paths The BFS enqueued every entry, so it held the whole subtree in memory including each file's chunk list. A filer serving a peer's first-time bootstrap traversal of a large tree could exhaust memory and get killed. Stream each entry as it is visited and queue only directory paths to descend into. Memory is now bounded by the number of directories rather than the entire tree, and the streamed output order is unchanged. * filer: match excluded prefixes on path-component boundaries Only treat an excluded prefix as a match when it ends at a path boundary, so excluding /a/b does not also drop a sibling like /a/bc. Short-circuit the trie walk on the first real match. --- .../server/filer_grpc_server_traverse_meta.go | 73 +++++++++++++------ .../filer_grpc_server_traverse_meta_test.go | 62 ++++++++++++++++ 2 files changed, 113 insertions(+), 22 deletions(-) diff --git a/weed/server/filer_grpc_server_traverse_meta.go b/weed/server/filer_grpc_server_traverse_meta.go index 9e317e83f..2ebf33d05 100644 --- a/weed/server/filer_grpc_server_traverse_meta.go +++ b/weed/server/filer_grpc_server_traverse_meta.go @@ -22,34 +22,63 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ ctx := stream.Context() - queue := util.NewQueue[*filer.Entry]() + isExcluded := func(path util.FullPath) bool { + excluded := false + excludedTrie.MatchPrefix([]byte(path), func(key []byte, value bool) bool { + // Only exclude when the prefix ends on a path-component boundary, + // so /a/b does not also exclude a sibling like /a/bc. + if len(path) == len(key) || path[len(key)] == '/' { + excluded = true + return false + } + return true + }) + return excluded + } + + // Send each entry as it is visited and queue only directory paths to + // descend into. Queuing the entries themselves would hold the whole + // subtree in memory, including every file's chunk list, and can exhaust + // the filer on large trees (e.g. during a peer's first-time bootstrap). + sendEntry := func(entry *filer.Entry) error { + parent, _ := entry.FullPath.DirAndName() + if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ + Directory: parent, + Entry: entry.ToProtoEntry(), + }); err != nil { + return fmt.Errorf("send traverse bfs metadata response: %w", err) + } + return nil + } + dirEntry, err := fs.filer.FindEntry(ctx, util.FullPath(req.Directory)) if err != nil { return fmt.Errorf("find dir %s: %v", req.Directory, err) } - queue.Enqueue(dirEntry) + if isExcluded(dirEntry.FullPath) { + return nil + } + if err := sendEntry(dirEntry); err != nil { + return err + } - for item := queue.Dequeue(); item != nil; item = queue.Dequeue() { - if excludedTrie.MatchPrefix([]byte(item.FullPath), func(key []byte, value bool) bool { - return true - }) { - // println("excluded", item.FullPath) - continue - } - parent, _ := item.FullPath.DirAndName() - if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ - Directory: parent, - Entry: item.ToProtoEntry(), - }); err != nil { - return fmt.Errorf("send traverse bfs metadata response: %w", err) - } + queue := util.NewQueue[util.FullPath]() + if dirEntry.IsDirectory() { + queue.Enqueue(dirEntry.FullPath) + } - if !item.IsDirectory() { - continue - } - - if err := fs.iterateDirectory(ctx, item.FullPath, func(entry *filer.Entry) error { - queue.Enqueue(entry) + for queue.Len() > 0 { + dirPath := queue.Dequeue() + if err := fs.iterateDirectory(ctx, dirPath, func(entry *filer.Entry) error { + if isExcluded(entry.FullPath) { + return nil + } + if err := sendEntry(entry); err != nil { + return err + } + if entry.IsDirectory() { + queue.Enqueue(entry.FullPath) + } return nil }); err != nil { return err diff --git a/weed/server/filer_grpc_server_traverse_meta_test.go b/weed/server/filer_grpc_server_traverse_meta_test.go index 31cd33064..7aa9e3a2e 100644 --- a/weed/server/filer_grpc_server_traverse_meta_test.go +++ b/weed/server/filer_grpc_server_traverse_meta_test.go @@ -1,10 +1,14 @@ package weed_server import ( + "context" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/assert" "github.com/viant/ptrie" + "google.golang.org/grpc" ) func TestPtrie(t *testing.T) { @@ -30,3 +34,61 @@ func TestPtrie(t *testing.T) { assert.False(t, excludedTrie.Has(b)) } + +type captureTraverseStream struct { + grpc.ServerStream + ctx context.Context + visited []string +} + +func (s *captureTraverseStream) Context() context.Context { return s.ctx } + +func (s *captureTraverseStream) Send(resp *filer_pb.TraverseBfsMetadataResponse) error { + dir := resp.Directory + if dir != "/" { + dir += "/" + } + s.visited = append(s.visited, dir+resp.Entry.Name) + return nil +} + +func TestTraverseBfsMetadata(t *testing.T) { + store := newRenameTestStore() + // "/.system-data" shares the "/.system" byte prefix but is a different + // path component, so it must not be excluded. + for _, dir := range []string{"/", "/a", "/a/sub", "/b", "/.system", "/.system-data"} { + store.entries[dir] = newDirectoryEntry(dir, 1) + } + for _, file := range []string{"/a/f1", "/a/sub/f2", "/b/f3", "/.system/secret", "/.system-data/keep"} { + store.entries[file] = newFileEntry(file, 2) + } + + server := &FilerServer{filer: newRenameTestFiler(store)} + stream := &captureTraverseStream{ctx: context.Background()} + + err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{ + Directory: "/", + ExcludedPrefixes: []string{"/.system"}, + }, stream) + assert.NoError(t, err) + + // The excluded subtree is skipped; everything else (including the + // /.system-data sibling) is visited exactly once. + assert.ElementsMatch(t, []string{ + "/", "/a", "/a/f1", "/a/sub", "/a/sub/f2", "/b", "/b/f3", + "/.system-data", "/.system-data/keep", + }, stream.visited) + + // Each entry's parent must be streamed before the entry itself, so a + // consumer can apply the tree top-down. + seen := make(map[string]bool) + for _, path := range stream.visited { + if path != "/" { + parent, _ := util.FullPath(path).DirAndName() + assert.Truef(t, seen[parent], "parent %s of %s not streamed first", parent, path) + } + seen[path] = true + } +} + +var _ filer_pb.SeaweedFiler_TraverseBfsMetadataServer = (*captureTraverseStream)(nil)