From 03f773f08728790f73a8a042d66a556a07a6b908 Mon Sep 17 00:00:00 2001 From: Junker der Provinz Date: Sun, 25 Oct 2026 00:04:37 +0200 Subject: [PATCH] redis2: pipeline entry reads when listing a directory (#12042) ListDirectoryEntries fetched every member with its own GET, so a page of 1024 entries cost 1024 round trips. Fetch the values in pipelined batches of 128 as the loop reaches them and walk the replies in order. Each reply keeps its own error, so missing and expired members are cleaned up and skipped exactly as before, and an unreadable member still ends the listing at that name. Batching keeps a caller that lists with a huge limit and stops after a few entries from reading the whole directory. Co-authored-by: Patrick Schmidt <1310072+Woellchen@users.noreply.github.com> Co-authored-by: user.email --- weed/filer/redis2/universal_redis_store.go | 36 +++++++- .../redis2/universal_redis_store_test.go | 89 +++++++++++++++++++ 2 files changed, 122 insertions(+), 3 deletions(-) diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go index 1320495583..2f90230e33 100644 --- a/weed/filer/redis2/universal_redis_store.go +++ b/weed/filer/redis2/universal_redis_store.go @@ -15,6 +15,10 @@ import ( const ( DIR_LIST_MARKER = "\x00" + + // a listing whose callback stops early reads at most one batch more than + // it uses, and no single pipeline grows with the listing limit + listFetchBatchSize = 128 ) type UniversalRedis2Store struct { @@ -96,7 +100,12 @@ func (store *UniversalRedis2Store) UpdateEntry(ctx context.Context, entry *filer func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) { - data, err := store.Client.Get(ctx, store.getKey(string(fullpath))).Result() + return decodeEntry(fullpath, store.Client.Get(ctx, store.getKey(string(fullpath)))) +} + +func decodeEntry(fullpath util.FullPath, get *redis.StringCmd) (entry *filer.Entry, err error) { + + data, err := get.Result() if err == redis.Nil { return nil, filer_pb.ErrNotFound } @@ -202,9 +211,13 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir // fetch entry meta var entry *filer.Entry - for _, fileName := range members { + var gets []*redis.StringCmd + for i, fileName := range members { + if i%listFetchBatchSize == 0 { + gets = store.getEntries(ctx, dirPath, members[i:]) + } path := util.NewFullPath(string(dirPath), fileName) - entry, err = store.FindEntry(ctx, path) + entry, err = decodeEntry(path, gets[i%listFetchBatchSize]) lastFileName = fileName if err != nil { glog.V(0).InfofCtx(ctx, "list %s : %v", path, err) @@ -235,6 +248,23 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir return lastFileName, err } +// getEntries pipelines the GETs for up to listFetchBatchSize of members. Each +// command keeps its own error, so a missing value still reads as redis.Nil and +// fails only that member. +func (store *UniversalRedis2Store) getEntries(ctx context.Context, dirPath util.FullPath, members []string) []*redis.StringCmd { + if len(members) > listFetchBatchSize { + members = members[:listFetchBatchSize] + } + gets := make([]*redis.StringCmd, len(members)) + store.Client.Pipelined(ctx, func(pipe redis.Pipeliner) error { + for i, fileName := range members { + gets[i] = pipe.Get(ctx, store.getKey(string(util.NewFullPath(string(dirPath), fileName)))) + } + return nil + }) + return gets +} + func (store *UniversalRedis2Store) removeOrphanedDirectoryListMember(ctx context.Context, dirPath util.FullPath, fileName string) { // a directory converted to super large after accumulating members still has a legacy index if store.isSuperLargeDirectory(string(dirPath)) { diff --git a/weed/filer/redis2/universal_redis_store_test.go b/weed/filer/redis2/universal_redis_store_test.go index 1558538dad..0bce68201c 100644 --- a/weed/filer/redis2/universal_redis_store_test.go +++ b/weed/filer/redis2/universal_redis_store_test.go @@ -256,6 +256,95 @@ func TestListDirectoryEntriesDeletesLogicallyExpiredEntries(t *testing.T) { } } +func TestListDirectoryEntriesSkipsMissingAndExpiredMembersInOrder(t *testing.T) { + for _, keyPrefix := range []string{"", "sw:"} { + t.Run("keyPrefix="+keyPrefix, func(t *testing.T) { + store, dir := newTestStore(t, keyPrefix) + + insertTestEntry(t, store, dir.Child("a"), 0) + insertTestEntry(t, store, dir.Child("b"), 0) + insertLogicallyExpiredTestEntry(t, store, dir.Child("c")) + insertTestEntry(t, store, dir.Child("d"), 0) + + if err := store.Client.Del(context.Background(), store.getKey(string(dir.Child("b")))).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + + if names := listNames(t, store, dir); fmt.Sprint(names) != "[a d]" { + t.Fatalf("listed %v, want [a d]", names) + } + + if members := indexMembers(t, store, dir); fmt.Sprint(members) != "[a d]" { + t.Fatalf("directory index holds %v, want [a d]", members) + } + }) + } +} + +func TestListDirectoryEntriesAcrossFetchBatches(t *testing.T) { + store, dir := newTestStore(t, "") + + var want []string + for i := 0; i < listFetchBatchSize+2; i++ { + name := fmt.Sprintf("f%04d", i) + insertTestEntry(t, store, dir.Child(name), 0) + if i != listFetchBatchSize { + want = append(want, name) + } + } + + // the first member of the second batch + missing := dir.Child(fmt.Sprintf("f%04d", listFetchBatchSize)) + if err := store.Client.Del(context.Background(), store.getKey(string(missing))).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + + var names []string + if _, err := store.ListDirectoryEntries(context.Background(), dir, "", true, 1000, func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }); err != nil { + t.Fatalf("ListDirectoryEntries %s: %v", dir, err) + } + if fmt.Sprint(names) != fmt.Sprint(want) { + t.Fatalf("listed %v, want %v", names, want) + } + + if members := indexMembers(t, store, dir); fmt.Sprint(members) != fmt.Sprint(want) { + t.Fatalf("directory index holds %v, want %v", members, want) + } +} + +func TestListDirectoryEntriesStopsAtUnreadableMember(t *testing.T) { + store, dir := newTestStore(t, "") + ctx := context.Background() + + insertTestEntry(t, store, dir.Child("a"), 0) + insertTestEntry(t, store, dir.Child("b"), 0) + insertTestEntry(t, store, dir.Child("c"), 0) + + // a value of the wrong type makes its GET fail with something other than redis.Nil + key := store.getKey(string(dir.Child("b"))) + if err := store.Client.Del(ctx, key).Err(); err != nil { + t.Fatalf("drop value key: %v", err) + } + if err := store.Client.HSet(ctx, key, "f", "v").Err(); err != nil { + t.Fatalf("plant hash: %v", err) + } + + var names []string + lastFileName, err := store.ListDirectoryEntries(ctx, dir, "", true, 100, func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err == nil { + t.Fatal("ListDirectoryEntries returned no error for an unreadable member") + } + if fmt.Sprint(names) != "[a]" || lastFileName != "b" { + t.Fatalf("listed %v up to %q, want [a] up to \"b\"", names, lastFileName) + } +} + func TestDeleteExpiredEntryKeepsRecreatedValue(t *testing.T) { store, dir := newTestStore(t, "")