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, "")