[#166] node: Parallelize background tree service sync

* Run sync task for nodes in parallel within errgroup worker pool

Signed-off-by: Airat Arifullin a.arifullin@yadro.com

pkg/services/tree/sync.go

@@ -17,6 +17,7 @@ import (
 	netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
 	"github.com/panjf2000/ants/v2"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -129,37 +130,58 @@ func (s *Service) synchronizeTree(ctx context.Context, cid cid.ID, from uint64,
 		zap.String("tree", treeID),
 		zap.Uint64("from", from))
 
-	newHeight := uint64(math.MaxUint64)
+	errGroup, egCtx := errgroup.WithContext(ctx)
-	for _, n := range nodes {
+	const workersCount = 4
-		height := from
+	errGroup.SetLimit(workersCount)
-		n.IterateNetworkEndpoints(func(addr string) bool {
-			var a network.Address
-			if err := a.FromString(addr); err != nil {
-				return false
-			}
 
-			cc, err := grpc.DialContext(ctx, a.URIAddr(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	heights := make([]uint64, len(nodes))
-			if err != nil {
+	for i, n := range nodes {
-				// Failed to connect, try the next address.
+		i := i
-				return false
+		n := n
-			}
+		errGroup.Go(func() error {
-			defer cc.Close()
+			height := from
+			n.IterateNetworkEndpoints(func(addr string) bool {
+				var a network.Address
+				if err := a.FromString(addr); err != nil {
+					return false
+				}
 
-			treeClient := NewTreeServiceClient(cc)
+				cc, err := grpc.DialContext(egCtx, a.URIAddr(), grpc.WithTransportCredentials(insecure.NewCredentials()))
-			for {
+				if err != nil {
-				h, err := s.synchronizeSingle(ctx, cid, treeID, height, treeClient)
+					// Failed to connect, try the next address.
-				if height < h {
+					return false
-					height = h
 				}
-				if err != nil || h <= height {
+				defer cc.Close()
-					// Error with the response, try the next node.
+
-					return true
+				treeClient := NewTreeServiceClient(cc)
+				for {
+					h, err := s.synchronizeSingle(egCtx, cid, treeID, height, treeClient)
+					if height < h {
+						height = h
+					}
+					if err != nil || h <= height {
+						// Error with the response, try the next node.
+						return true
+					}
 				}
+			})
+
+			if height <= from { // do not increase starting height on fail
+				heights[i] = from
+				return nil
 			}
+			heights[i] = height
+			return nil
 		})
-		if height <= from { // do not increase starting height on fail
+	}
-			newHeight = from
+
-		} else if height < newHeight { // take minimum across all clients
+	if err := errGroup.Wait(); err != nil {
+		s.log.Warn("failed to run tree synchronization over all nodes", zap.Error(err))
+	}
+
+	newHeight := uint64(math.MaxUint64)
+	for _, height := range heights { // take minimum across all clients
+		if height < newHeight {
 			newHeight = height
 		}
 	}