share: handle rows concurrently

CHANGELOG-PENDING.md

@@ -23,6 +23,7 @@ Month, DD, YYYY
 - [feat(node): extract overrides from Config into Settings #292](https://github.com/celestiaorg/celestia-node/pull/292) [@Wondertan](https://github.com/Wondertan)
 
 - [docker] Created `docker/` dir with `Dockerfile` and `entrypoint.sh` script. 
+- [chore(share): handle rows concurrently in GetSharesByNamespace #241](https://github.com/celestiaorg/celestia-node/pull/241) [@vgonkivs](https://github.com/vgonkivs)
 
 ### BUG FIXES
 

Makefile

@@ -51,6 +51,12 @@ test:
 	@go test -v ./...
 .PHONY: test
 
+## benchmark: Running all benchmarks
+benchmark:
+	@echo "--> Running benchmarks"
+	@go test -run="none" -bench=. -benchtime=100x -benchmem ./...
+.PHONY: benchmark
+
 PB_PKGS=$(shell find . -name 'pb' -type d)
 PB_CORE=$(shell go list -f {{.Dir}} -m github.com/tendermint/tendermint)
 PB_GOGO=$(shell go list -f {{.Dir}} -m github.com/gogo/protobuf)

service/share/share.go

@@ -144,17 +144,37 @@ func (s *service) GetSharesByNamespace(ctx context.Context, root *Root, nID name
 		return nil, ipld.ErrNotFoundInRange
 	}
 
-	namespacedShares := make([]Share, 0)
+	type res struct {
+		nodes []format.Node
+		err   error
+	}
+	resultCh := make(chan *res)
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	for _, rootCID := range rowRootCIDs {
-		nodes, err := ipld.GetLeavesByNamespace(ctx, s.dag, rootCID, nID)
-		if err != nil {
-			return nil, err
-		}
+		go func(rootCID cid.Cid) {
+			nodes, err := ipld.GetLeavesByNamespace(ctx, s.dag, rootCID, nID)
+			resultCh <- &res{nodes: nodes, err: err}
+		}(rootCID)
+	}
 
-		for _, node := range nodes {
-			namespacedShares = append(namespacedShares, node.RawData()[1:])
+	namespacedShares := make([]Share, 0)
+	for i := 0; i < len(rowRootCIDs); i++ {
+		select {
+		case result := <-resultCh:
+			if result.err != nil {
+				return nil, result.err
+			}
+			for _, node := range result.nodes {
+				namespacedShares = append(namespacedShares, node.RawData()[1:])
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
 		}
 	}
+
 	return namespacedShares, nil
 }
 

service/share/share_test.go

@@ -52,7 +52,42 @@ func TestService_GetSharesByNamespace(t *testing.T) {
 			shares, err := serv.GetSharesByNamespace(context.Background(), root, randNID)
 			require.NoError(t, err)
 			assert.Len(t, shares, tt.expectedShareCount)
-			assert.Equal(t, randNID, []byte(shares[0].NamespaceID()))
+			for _, value := range shares {
+				assert.Equal(t, randNID, []byte(value.NamespaceID()))
+			}
+		})
+	}
+}
+
+func TestService_GetSharesByNamespaceNotFoundInRange(t *testing.T) {
+	serv, root := RandServiceWithSquare(t, 1)
+	root.RowsRoots = nil
+
+	shares, err := serv.GetSharesByNamespace(context.Background(), root, []byte(""))
+	assert.Len(t, shares, 0)
+	require.Error(t, err, "namespaceID not found in range")
+}
+
+func BenchmarkService_GetSharesByNamespace(b *testing.B) {
+	var tests = []struct {
+		amountShares int
+	}{
+		{amountShares: 4},
+		{amountShares: 16},
+		{amountShares: 128},
+	}
+
+	for _, tt := range tests {
+		b.Run(strconv.Itoa(tt.amountShares), func(b *testing.B) {
+			t := &testing.T{}
+			serv, root := RandServiceWithSquare(t, tt.amountShares)
+			randNID := root.RowsRoots[(len(root.RowsRoots)-1)/2][:8]
+			root.RowsRoots[(len(root.RowsRoots) / 2)] = root.RowsRoots[(len(root.RowsRoots)-1)/2]
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, err := serv.GetSharesByNamespace(context.Background(), root, randNID)
+				require.NoError(t, err)
+			}
 		})
 	}
 }