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 <chrislusf@users.noreply.github.com>

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)) {

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