core: Reduce rebroadcast BA result (#433)

* core: reduce processing duplicated agreement result

* 放這邊應該比較好

* 放這邊才對

* Add TODO

2 files changed, 36 insertions, 3 deletions

diff --git a/core/agreement-mgr.go b/core/agreement-mgr.go
index a8fab7c..e07b23f 100644
--- a/core/agreement-mgr.go
+++ b/core/agreement-mgr.go
@@ -94,6 +94,7 @@ type agreementMgr struct {
 	initRound     uint64
 	configs       []*agreementMgrConfig
 	baModules     []*agreement
+	lastBAResult  []types.Position
 	voteFilters   []*utils.VoteFilter
 	waitGroup     sync.WaitGroup
 	pendingVotes  map[uint64][]*types.Vote
@@ -203,6 +204,10 @@ func (mgr *agreementMgr) appendConfig(
 		recv.agreementModule = agrModule
 		mgr.baModules = append(mgr.baModules, agrModule)
 		mgr.voteFilters = append(mgr.voteFilters, utils.NewVoteFilter())
+		mgr.lastBAResult = append(mgr.lastBAResult, types.Position{
+			Round:   round,
+			ChainID: i,
+		})
 		if mgr.isRunning {
 			mgr.waitGroup.Add(1)
 			go func(idx uint32) {
@@ -251,6 +256,21 @@ func (mgr *agreementMgr) processBlock(b *types.Block) error {
 	return mgr.baModules[b.Position.ChainID].processBlock(b)
 }
 
+func (mgr *agreementMgr) firstAgreementResult(
+	result *types.AgreementResult) (bool, error) {
+	mgr.lock.RLock()
+	defer mgr.lock.RUnlock()
+	if result.Position.ChainID >= uint32(len(mgr.lastBAResult)) {
+		mgr.logger.Error("Process unknown result for unknown chain to BA",
+			"position", &result.Position,
+			"baChain", len(mgr.lastBAResult),
+			"baRound", len(mgr.configs),
+			"initRound", mgr.initRound)
+		return false, utils.ErrInvalidChainID
+	}
+	return result.Position.Newer(&mgr.lastBAResult[result.Position.ChainID]), nil
+}
+
 func (mgr *agreementMgr) processAgreementResult(
 	result *types.AgreementResult) error {
 	mgr.lock.RLock()
@@ -263,6 +283,10 @@ func (mgr *agreementMgr) processAgreementResult(
 			"initRound", mgr.initRound)
 		return utils.ErrInvalidChainID
 	}
+	// TODO(jimmy): lock in this function is not safe.
+	if result.Position.Newer(&mgr.lastBAResult[result.Position.ChainID]) {
+		mgr.lastBAResult[result.Position.ChainID] = result.Position
+	}
 	agreement := mgr.baModules[result.Position.ChainID]
 	aID := agreement.agreementID()
 	if isStop(aID) {
diff --git a/core/consensus.go b/core/consensus.go
index 223e5df..623ad2b 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -1026,6 +1026,12 @@ func (con *Consensus) ProcessAgreementResult(
 		return err
 	}
 	con.lattice.AddShallowBlock(rand.BlockHash, rand.Position)
+
+	first, err := con.baMgr.firstAgreementResult(rand)
+	if err != nil {
+		return err
+	}
+
 	// Syncing BA Module.
 	if err := con.baMgr.processAgreementResult(rand); err != nil {
 		return err
@@ -1039,9 +1045,12 @@ func (con *Consensus) ProcessAgreementResult(
 	if !con.cfgModule.touchTSigHash(rand.BlockHash) {
 		return nil
 	}
-	con.logger.Debug("Rebroadcast AgreementResult",
-		"result", rand)
-	con.network.BroadcastAgreementResult(rand)
+
+	if first {
+		con.logger.Debug("Rebroadcast AgreementResult",
+			"result", rand)
+		con.network.BroadcastAgreementResult(rand)
+	}
 	dkgSet, err := con.nodeSetCache.GetDKGSet(rand.Position.Round)
 	if err != nil {
 		return err