refactor: optimized GetSortedRevealedValues (#1173)

* refactor: fetched reveal event logs concurrently

* refactor: calculated GetSortedRevealedValues() concurrently

cmd/propose.go

@@ -391,45 +391,92 @@ func (*UtilsStruct) GetSortedRevealedValues(client *ethclient.Client, blockNumbe
 		return nil, err
 	}
 	log.Debugf("GetSortedRevealedValues: Revealed Data: %+v", assignedAsset)
+
+	var wg sync.WaitGroup
+	resultsChan := make(chan *types.AssetResult, len(assignedAsset))
+
+	for _, asset := range assignedAsset {
+		wg.Add(1)
+		go processAsset(asset, resultsChan, &wg)
+	}
+
+	wg.Wait()
+	close(resultsChan)
+
 	revealedValuesWithIndex := make(map[uint16][]*big.Int)
 	voteWeights := make(map[string]*big.Int)
 	influenceSum := make(map[uint16]*big.Int)
-	log.Debug("Calculating sorted revealed values, vote weights and influence sum...")
-	for _, asset := range assignedAsset {
-		for _, assetValue := range asset.RevealedValues {
-			if revealedValuesWithIndex[assetValue.LeafId] == nil {
-				revealedValuesWithIndex[assetValue.LeafId] = []*big.Int{assetValue.Value}
+
+	for result := range resultsChan {
+		for leafId, values := range result.RevealedValuesWithIndex {
+			revealedValuesWithIndex[leafId] = append(revealedValuesWithIndex[leafId], values...)
+		}
+		for value, weight := range result.VoteWeights {
+			if voteWeights[value] == nil {
+				voteWeights[value] = weight
 			} else {
-				if !utils.ContainsBigInteger(revealedValuesWithIndex[assetValue.LeafId], assetValue.Value) {
-					revealedValuesWithIndex[assetValue.LeafId] = append(revealedValuesWithIndex[assetValue.LeafId], assetValue.Value)
-				}
+				voteWeights[value].Add(voteWeights[value], weight)
 			}
-			//Calculate vote weights
-			if voteWeights[assetValue.Value.String()] == nil {
-				voteWeights[assetValue.Value.String()] = big.NewInt(0)
-			}
-			voteWeights[assetValue.Value.String()] = big.NewInt(0).Add(voteWeights[assetValue.Value.String()], asset.Influence)
-
-			//Calculate influence sum
-			if influenceSum[assetValue.LeafId] == nil {
-				influenceSum[assetValue.LeafId] = big.NewInt(0)
+		}
+		for leafId, sum := range result.InfluenceSum {
+			if influenceSum[leafId] == nil {
+				influenceSum[leafId] = sum
+			} else {
+				influenceSum[leafId].Add(influenceSum[leafId], sum)
 			}
-			influenceSum[assetValue.LeafId] = big.NewInt(0).Add(influenceSum[assetValue.LeafId], asset.Influence)
 		}
 	}
-	//sort revealed values
-	for _, element := range revealedValuesWithIndex {
-		sort.Slice(element, func(i, j int) bool {
-			return element[i].Cmp(element[j]) == -1
+
+	for _, values := range revealedValuesWithIndex {
+		sort.Slice(values, func(i, j int) bool {
+			return values[i].Cmp(values[j]) == -1
 		})
 	}
+
 	return &types.RevealedDataMaps{
 		SortedRevealedValues: revealedValuesWithIndex,
 		VoteWeights:          voteWeights,
 		InfluenceSum:         influenceSum,
 	}, nil
 }
 
+func processAsset(asset types.RevealedStruct, resultsChan chan<- *types.AssetResult, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	revealedValuesWithIndex := make(map[uint16][]*big.Int)
+	voteWeights := make(map[string]*big.Int)
+	influenceSum := make(map[uint16]*big.Int)
+
+	for _, assetValue := range asset.RevealedValues {
+		leafId := assetValue.LeafId
+		valueStr := assetValue.Value.String()
+		influence := asset.Influence
+
+		// Append the leaf value to the revealed values slice if it's not already present
+		if !utils.ContainsBigInteger(revealedValuesWithIndex[leafId], assetValue.Value) {
+			revealedValuesWithIndex[leafId] = append(revealedValuesWithIndex[leafId], assetValue.Value)
+		}
+
+		// Calculate vote weights
+		if voteWeights[valueStr] == nil {
+			voteWeights[valueStr] = big.NewInt(0)
+		}
+		voteWeights[valueStr] = voteWeights[valueStr].Add(voteWeights[valueStr], influence)
+
+		// Calculate influence sum
+		if influenceSum[leafId] == nil {
+			influenceSum[leafId] = big.NewInt(0)
+		}
+		influenceSum[leafId] = influenceSum[leafId].Add(influenceSum[leafId], influence)
+	}
+
+	resultsChan <- &types.AssetResult{
+		RevealedValuesWithIndex: revealedValuesWithIndex,
+		VoteWeights:             voteWeights,
+		InfluenceSum:            influenceSum,
+	}
+}
+
 //This function returns the medians, idsRevealedInThisEpoch and revealedDataMaps
 func (*UtilsStruct) MakeBlock(client *ethclient.Client, blockNumber *big.Int, epoch uint32, rogueData types.Rogue) ([]*big.Int, []uint16, *types.RevealedDataMaps, error) {
 	log.Debugf("MakeBlock: Calling GetSortedRevealedValues with arguments blockNumber = %s, epoch = %d", blockNumber, epoch)

cmd/reveal.go

@@ -9,9 +9,11 @@ import (
 	"razor/pkg/bindings"
 	"razor/utils"
 	"strings"
+	"sync"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
+	Types "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
 )
 
@@ -135,14 +137,25 @@ func (*UtilsStruct) IndexRevealEventsOfCurrentEpoch(client *ethclient.Client, bl
 	if err != nil {
 		return nil, err
 	}
-	var revealedData []types.RevealedStruct
+
+	revealedDataChan := make(chan types.RevealedStruct, len(logs))
+	errorChan := make(chan error, len(logs))
+	var wg sync.WaitGroup
+
 	for _, vLog := range logs {
-		data, unpackErr := abiUtils.Unpack(contractAbi, "Revealed", vLog.Data)
-		if unpackErr != nil {
-			log.Debug(unpackErr)
-			continue
-		}
-		if epoch == data[0].(uint32) {
+		wg.Add(1)
+		go func(vLog Types.Log) {
+			defer wg.Done()
+			data, unpackErr := abiUtils.Unpack(contractAbi, "Revealed", vLog.Data)
+			if unpackErr != nil {
+				errorChan <- unpackErr
+				return
+			}
+
+			if epoch != data[0].(uint32) {
+				return
+			}
+
 			treeValues := data[2].([]struct {
 				LeafId uint16   `json:"leafId"`
 				Value  *big.Int `json:"value"`
@@ -158,9 +171,31 @@ func (*UtilsStruct) IndexRevealEventsOfCurrentEpoch(client *ethclient.Client, bl
 				RevealedValues: revealedValues,
 				Influence:      data[1].(*big.Int),
 			}
-			revealedData = append(revealedData, consolidatedRevealedData)
+
+			revealedDataChan <- consolidatedRevealedData
+		}(vLog)
+	}
+
+	wg.Wait()
+	close(revealedDataChan)
+	close(errorChan)
+
+	var revealedData []types.RevealedStruct
+	for {
+		select {
+		case data, ok := <-revealedDataChan:
+			if ok {
+				revealedData = append(revealedData, data)
+			}
+		case err, ok := <-errorChan:
+			if ok {
+				log.Debug(err)
+			}
+		}
+		if len(revealedDataChan) == 0 && len(errorChan) == 0 {
+			break
 		}
 	}
-	log.Debug("IndexRevealEventsOfCurrentEpoch: Revealed values: ", revealedData)
+
 	return revealedData, nil
 }

core/types/assets.go

@@ -29,6 +29,12 @@ type Asset struct {
 	Collection bindings.StructsCollection
 }
 
+type AssetResult struct {
+	RevealedValuesWithIndex map[uint16][]*big.Int
+	VoteWeights             map[string]*big.Int
+	InfluenceSum            map[uint16]*big.Int
+}
+
 type Locks struct {
 	Amount      *big.Int
 	UnlockAfter *big.Int