Added HandleMessageUpdate locks.

consensus/consensus.go

@@ -200,7 +200,7 @@ func (consensus *Consensus) GetPrivateKeys() multibls.PrivateKeys {
 }
 
 // GetLeaderPrivateKey returns leader private key if node is the leader
-func (consensus *Consensus) GetLeaderPrivateKey(leaderKey *bls_core.PublicKey) (*bls.PrivateKeyWrapper, error) {
+func (consensus *Consensus) getLeaderPrivateKey(leaderKey *bls_core.PublicKey) (*bls.PrivateKeyWrapper, error) {
 	for i, key := range consensus.priKey {
 		if key.Pub.Object.IsEqual(leaderKey) {
 			return &consensus.priKey[i], nil
@@ -209,9 +209,9 @@ func (consensus *Consensus) GetLeaderPrivateKey(leaderKey *bls_core.PublicKey) (
 	return nil, errors.Wrapf(errLeaderPriKeyNotFound, leaderKey.SerializeToHexStr())
 }
 
-// GetConsensusLeaderPrivateKey returns consensus leader private key if node is the leader
-func (consensus *Consensus) GetConsensusLeaderPrivateKey() (*bls.PrivateKeyWrapper, error) {
-	return consensus.GetLeaderPrivateKey(consensus.LeaderPubKey.Object)
+// getConsensusLeaderPrivateKey returns consensus leader private key if node is the leader
+func (consensus *Consensus) getConsensusLeaderPrivateKey() (*bls.PrivateKeyWrapper, error) {
+	return consensus.getLeaderPrivateKey(consensus.LeaderPubKey.Object)
 }
 
 // SetBlockVerifier sets the block verifier

consensus/consensus_service.go

@@ -468,7 +468,12 @@ func (consensus *Consensus) isLeader() bool {
 // SetViewIDs set both current view ID and view changing ID to the height
 // of the blockchain. It is used during client startup to recover the state
 func (consensus *Consensus) SetViewIDs(height uint64) {
-	fmt.Println("SetViewIDs", height)
+	consensus.setViewIDs(height)
+}
+
+// SetViewIDs set both current view ID and view changing ID to the height
+// of the blockchain. It is used during client startup to recover the state
+func (consensus *Consensus) setViewIDs(height uint64) {
 	consensus.SetCurBlockViewID(height)
 	consensus.SetViewChangingID(height)
 }

consensus/consensus_v2.go

@@ -56,6 +56,8 @@ func (consensus *Consensus) isViewChangingMode() bool {
 
 // HandleMessageUpdate will update the consensus state according to received message
 func (consensus *Consensus) HandleMessageUpdate(ctx context.Context, msg *msg_pb.Message, senderKey *bls.SerializedPublicKey) error {
+	consensus.mutex.Lock()
+	defer consensus.mutex.Unlock()
 	// when node is in ViewChanging mode, it still accepts normal messages into FBFTLog
 	// in order to avoid possible trap forever but drop PREPARE and COMMIT
 	// which are message types specifically for a node acting as leader
@@ -144,7 +146,7 @@ func (consensus *Consensus) finalCommit() {
 		Msg("[finalCommit] Finalizing Consensus")
 	beforeCatchupNum := consensus.BlockNum()
 
-	leaderPriKey, err := consensus.GetConsensusLeaderPrivateKey()
+	leaderPriKey, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
 		consensus.getLogger().Error().Err(err).Msg("[finalCommit] leader not found")
 		return
@@ -559,7 +561,7 @@ func (consensus *Consensus) preCommitAndPropose(blk *types.Block) error {
 		return errors.New("block to pre-commit is nil")
 	}
 
-	leaderPriKey, err := consensus.GetConsensusLeaderPrivateKey()
+	leaderPriKey, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
 		consensus.getLogger().Error().Err(err).Msg("[preCommitAndPropose] leader not found")
 		return err
@@ -804,7 +806,7 @@ func (consensus *Consensus) postCatchup(initBN uint64) {
 
 // GenerateVrfAndProof generates new VRF/Proof from hash of previous block
 func (consensus *Consensus) GenerateVrfAndProof(newHeader *block.Header) error {
-	key, err := consensus.GetConsensusLeaderPrivateKey()
+	key, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
 		return errors.New("[GenerateVrfAndProof] no leader private key provided")
 	}

consensus/leader.go

@@ -29,7 +29,7 @@ func (consensus *Consensus) announce(block *types.Block) {
 	copy(consensus.blockHash[:], blockHash[:])
 	consensus.block = encodedBlock // Must set block bytes before consensus.construct()
 
-	key, err := consensus.GetConsensusLeaderPrivateKey()
+	key, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
 		consensus.getLogger().Warn().Err(err).Msg("[Announce] Node not a leader")
 		return
@@ -102,11 +102,6 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 			Uint64("MsgBlockNum", recvMsg.BlockNum).
 			Msg("[OnPrepare] No Matching Announce message")
 	}
-
-	//// Read - Start
-	consensus.mutex.Lock()
-	defer consensus.mutex.Unlock()
-
 	if !consensus.isRightBlockNumAndViewID(recvMsg) {
 		return
 	}
@@ -195,8 +190,6 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 }
 
 func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
-	consensus.mutex.Lock()
-	defer consensus.mutex.Unlock()
 	//// Read - Start
 	if !consensus.isRightBlockNumAndViewID(recvMsg) {
 		return

consensus/threshold.go

@@ -15,7 +15,7 @@ import (
 func (consensus *Consensus) didReachPrepareQuorum() error {
 	logger := utils.Logger()
 	logger.Info().Msg("[OnPrepare] Received Enough Prepare Signatures")
-	leaderPriKey, err := consensus.GetConsensusLeaderPrivateKey()
+	leaderPriKey, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
 		utils.Logger().Warn().Err(err).Msg("[OnPrepare] leader not found")
 		return err

consensus/view_change.go

@@ -347,10 +347,6 @@ func (consensus *Consensus) startNewView(viewID uint64, newLeaderPriKey *bls.Pri
 
 // onViewChange is called when the view change message is received.
 func (consensus *Consensus) onViewChange(recvMsg *FBFTMessage) {
-	//fmt.Printf("[onViewChange] received view change message from %+v\n", recvMsg)
-	consensus.mutex.Lock()
-	defer consensus.mutex.Unlock()
-
 	consensus.getLogger().Debug().
 		Uint64("viewID", recvMsg.ViewID).
 		Uint64("blockNum", recvMsg.BlockNum).
@@ -359,7 +355,7 @@ func (consensus *Consensus) onViewChange(recvMsg *FBFTMessage) {
 
 	// if not leader, noop
 	newLeaderKey := recvMsg.LeaderPubkey
-	newLeaderPriKey, err := consensus.GetLeaderPrivateKey(newLeaderKey.Object)
+	newLeaderPriKey, err := consensus.getLeaderPrivateKey(newLeaderKey.Object)
 	if err != nil {
 		consensus.getLogger().Debug().
 			Err(err).
@@ -454,9 +450,6 @@ func (consensus *Consensus) onViewChange(recvMsg *FBFTMessage) {
 // Or the validator will enter announce phase to wait for the new block proposed
 // from the new leader
 func (consensus *Consensus) onNewView(recvMsg *FBFTMessage) {
-	consensus.mutex.Lock()
-	defer consensus.mutex.Unlock()
-
 	consensus.getLogger().Info().
 		Uint64("viewID", recvMsg.ViewID).
 		Uint64("blockNum", recvMsg.BlockNum).
@@ -543,7 +536,7 @@ func (consensus *Consensus) onNewView(recvMsg *FBFTMessage) {
 	consensus.consensusTimeout[timeoutViewChange].Stop()
 
 	// newView message verified success, override my state
-	consensus.SetViewIDs(recvMsg.ViewID)
+	consensus.setViewIDs(recvMsg.ViewID)
 	consensus.LeaderPubKey = senderKey
 	consensus.ResetViewChangeState()