PoS_EVM
/
woop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[viewchange] encapsulate both viewID and viewChangeID

Browse Source 
[viewchange] encapsulate view ID in the State struct

do NOT ues consensus.current directory to set/get viewID

use the following functions

consensus.SetCurViewID, consensus.SetViewChangingID
consensus.GetCurViewID, consensus.GetViewChangingID

Signed-off-by: Leo Chen <leo@harmony.one>
pull/3347/head
Leo Chen 4 years ago
parent c2e962a035
commit 42923336c9
14 changed files with 146 additions and 121 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 27
      consensus/checks.go
  2. 3
      consensus/consensus.go
  3. 52
      consensus/consensus_service.go
  4. 41
      consensus/consensus_v2.go
  5. 6
      consensus/consensus_viewchange_msg.go
  6. 2
      consensus/construct.go
  7. 12
      consensus/debug.go
  8. 5
      consensus/leader.go
  9. 97
      consensus/view_change.go
  10. 4
      consensus/view_change_test.go
  11. 2
      hmy/hmy.go
  12. 10
      node/api.go
  13. 2
      node/node_newblock.go
  14. 4
      rpc/debug.go

27
consensus/checks.go
Unescape View File

@ -57,7 +57,7 @@ func (consensus *Consensus) senderKeySanityChecks(msg *msg_pb.Message, senderKey
func (consensus *Consensus) isRightBlockNumAndViewID(recvMsg *FBFTMessage, func (consensus *Consensus) isRightBlockNumAndViewID(recvMsg *FBFTMessage,
) bool { ) bool {
if recvMsg.ViewID != consensus.viewID || recvMsg.BlockNum != consensus.blockNum { if recvMsg.ViewID != consensus.GetCurViewID() || recvMsg.BlockNum != consensus.blockNum {
consensus.getLogger().Debug(). consensus.getLogger().Debug().
Uint64("MsgViewID", recvMsg.ViewID). Uint64("MsgViewID", recvMsg.ViewID).
Uint64("MsgBlockNum", recvMsg.BlockNum). Uint64("MsgBlockNum", recvMsg.BlockNum).
@ -96,7 +96,7 @@ func (consensus *Consensus) onAnnounceSanityChecks(recvMsg *FBFTMessage) bool {
"[OnAnnounce] Already in ViewChanging mode, conflicing announce, doing noop", "[OnAnnounce] Already in ViewChanging mode, conflicing announce, doing noop",
) )
} else { } else {
consensus.startViewChange(consensus.viewID + 1) consensus.startViewChange(consensus.GetCurViewID() + 1)
} }
} }
consensus.getLogger().Debug(). consensus.getLogger().Debug().
@ -162,43 +162,46 @@ func (consensus *Consensus) onPreparedSanityChecks(
return true return true
} }
// TODO: leo: move the sanity check to p2p message validation
func (consensus *Consensus) onViewChangeSanityCheck(recvMsg *FBFTMessage) bool { func (consensus *Consensus) onViewChangeSanityCheck(recvMsg *FBFTMessage) bool {
// TODO: if difference is only one, new leader can still propose the same committed block to avoid another view change // TODO: if difference is only one, new leader can still propose the same committed block to avoid another view change
// TODO: new leader catchup without ignore view change message // TODO: new leader catchup without ignore view change message
consensus.getLogger().Info().
Uint64("MsgBlockNum", recvMsg.BlockNum).
Uint64("MyViewChangingID", consensus.GetViewChangingID()).
Uint64("MsgViewChangingID", recvMsg.ViewID).
Msg("onViewChange")
if consensus.blockNum > recvMsg.BlockNum { if consensus.blockNum > recvMsg.BlockNum {
consensus.getLogger().Debug(). consensus.getLogger().Debug().
Uint64("MsgBlockNum", recvMsg.BlockNum).
Msg("[onViewChange] Message BlockNum Is Low") Msg("[onViewChange] Message BlockNum Is Low")
return false return false
} }
if consensus.blockNum < recvMsg.BlockNum { if consensus.blockNum < recvMsg.BlockNum {
consensus.getLogger().Warn(). consensus.getLogger().Warn().
Uint64("MsgBlockNum", recvMsg.BlockNum).
Msg("[onViewChange] New Leader Has Lower Blocknum") Msg("[onViewChange] New Leader Has Lower Blocknum")
return false return false
} }
if consensus.IsViewChangingMode() && if consensus.IsViewChangingMode() &&
consensus.current.ViewID() > recvMsg.ViewID { consensus.GetViewChangingID() > recvMsg.ViewID {
consensus.getLogger().Warn(). consensus.getLogger().Warn().
Uint64("MyViewChangingID", consensus.current.ViewID()).
Uint64("MsgViewChangingID", recvMsg.ViewID).
Msg("[onViewChange] ViewChanging ID Is Low") Msg("[onViewChange] ViewChanging ID Is Low")
return false return false
} }
if recvMsg.ViewID-consensus.current.ViewID() > MaxViewIDDiff { if recvMsg.ViewID-consensus.GetViewChangingID() > MaxViewIDDiff {
consensus.getLogger().Debug(). consensus.getLogger().Debug().
Uint64("MsgViewID", recvMsg.ViewID).
Uint64("CurrentViewID", consensus.current.ViewID()).
Msg("Received viewID that is MaxViewIDDiff (100) further from the current viewID!") Msg("Received viewID that is MaxViewIDDiff (100) further from the current viewID!")
return false return false
} }
return true return true
} }
// TODO: leo: move the sanity check to p2p message validation
func (consensus *Consensus) onNewViewSanityCheck(recvMsg *FBFTMessage) bool { func (consensus *Consensus) onNewViewSanityCheck(recvMsg *FBFTMessage) bool {
if recvMsg.ViewID < consensus.viewID { if recvMsg.ViewID < consensus.GetCurViewID() {
consensus.getLogger().Warn(). consensus.getLogger().Warn().
Uint64("LastSuccessfulConsensusViewID", consensus.viewID). Uint64("LastSuccessfulConsensusViewID", consensus.GetCurViewID()).
Uint64("MsgViewChangingID", recvMsg.ViewID). Uint64("MsgViewChangingID", recvMsg.ViewID).
Msg("[onNewView] ViewID should be larger than the viewID of the last successful consensus") Msg("[onNewView] ViewID should be larger than the viewID of the last successful consensus")
return false return false

3
consensus/consensus.go
Unescape View File

@ -77,7 +77,6 @@ type Consensus struct {
// blockNum: the next blockNumber that FBFT is going to agree on, // blockNum: the next blockNumber that FBFT is going to agree on,
// should be equal to the blockNumber of next block // should be equal to the blockNumber of next block
blockNum uint64 blockNum uint64
viewID uint64
// Blockhash - 32 byte // Blockhash - 32 byte
blockHash [32]byte blockHash [32]byte
// Block to run consensus on // Block to run consensus on
@ -199,7 +198,7 @@ func New(
// viewID has to be initialized as the height of // viewID has to be initialized as the height of
// the blockchain during initialization as it was // the blockchain during initialization as it was
// displayed on explorer as Height right now // displayed on explorer as Height right now
consensus.viewID = 0 consensus.SetCurViewID(0)
consensus.ShardID = shard consensus.ShardID = shard
consensus.syncReadyChan = make(chan struct{}) consensus.syncReadyChan = make(chan struct{})
consensus.syncNotReadyChan = make(chan struct{}) consensus.syncNotReadyChan = make(chan struct{})

52
consensus/consensus_service.go
Unescape View File

@ -72,11 +72,6 @@ func (consensus *Consensus) signAndMarshalConsensusMessage(message *msg_pb.Messa
return marshaledMessage, nil return marshaledMessage, nil
} }
// GetViewID returns the consensus ID
func (consensus *Consensus) GetViewID() uint64 {
return consensus.viewID
}
// UpdatePublicKeys updates the PublicKeys for // UpdatePublicKeys updates the PublicKeys for
// quorum on current subcommittee, protected by a mutex // quorum on current subcommittee, protected by a mutex
func (consensus *Consensus) UpdatePublicKeys(pubKeys []bls_cosi.PublicKeyWrapper) int64 { func (consensus *Consensus) UpdatePublicKeys(pubKeys []bls_cosi.PublicKeyWrapper) int64 {
@ -199,12 +194,6 @@ func (consensus *Consensus) IsValidatorInCommittee(pubKey bls.SerializedPublicKe
return consensus.Decider.IndexOf(pubKey) != -1 return consensus.Decider.IndexOf(pubKey) != -1
} }
// SetViewID set the viewID to the height of the blockchain
func (consensus *Consensus) SetViewID(height uint64) {
consensus.viewID = height
consensus.current.viewID = height
}
// SetMode sets the mode of consensus // SetMode sets the mode of consensus
func (consensus *Consensus) SetMode(m Mode) { func (consensus *Consensus) SetMode(m Mode) {
consensus.current.SetMode(m) consensus.current.SetMode(m)
@ -232,26 +221,20 @@ func (consensus *Consensus) checkViewID(msg *FBFTMessage) error {
//in syncing mode, node accepts incoming messages without viewID/leaderKey checking //in syncing mode, node accepts incoming messages without viewID/leaderKey checking
//so only set mode to normal when new node enters consensus and need checking viewID //so only set mode to normal when new node enters consensus and need checking viewID
consensus.current.SetMode(Normal) consensus.current.SetMode(Normal)
consensus.viewID = msg.ViewID consensus.SetViewID(msg.ViewID)
consensus.current.SetViewID(msg.ViewID)
if len(msg.SenderPubkeys) != 1 { if len(msg.SenderPubkeys) != 1 {
return errors.New("Leader message can not have multiple sender keys") return errors.New("Leader message can not have multiple sender keys")
} }
consensus.LeaderPubKey = msg.SenderPubkeys[0] consensus.LeaderPubKey = msg.SenderPubkeys[0]
consensus.IgnoreViewIDCheck.UnSet() consensus.IgnoreViewIDCheck.UnSet()
consensus.consensusTimeout[timeoutConsensus].Start() consensus.consensusTimeout[timeoutConsensus].Start()
utils.Logger().Debug(). consensus.getLogger().Debug().
Uint64("viewID", consensus.viewID).
Str("leaderKey", consensus.LeaderPubKey.Bytes.Hex()). Str("leaderKey", consensus.LeaderPubKey.Bytes.Hex()).
Msg("viewID and leaderKey override")
utils.Logger().Debug().
Uint64("viewID", consensus.viewID).
Uint64("block", consensus.blockNum).
Msg("Start consensus timer") Msg("Start consensus timer")
return nil return nil
} else if msg.ViewID > consensus.viewID { } else if msg.ViewID > consensus.GetCurViewID() {
return consensus_engine.ErrViewIDNotMatch return consensus_engine.ErrViewIDNotMatch
} else if msg.ViewID < consensus.viewID { } else if msg.ViewID < consensus.GetCurViewID() {
return errors.New("view ID belongs to the past") return errors.New("view ID belongs to the past")
} }
return nil return nil
@ -282,7 +265,7 @@ func (consensus *Consensus) ReadSignatureBitmapPayload(
func (consensus *Consensus) getLogger() *zerolog.Logger { func (consensus *Consensus) getLogger() *zerolog.Logger {
logger := utils.Logger().With(). logger := utils.Logger().With().
Uint64("myBlock", consensus.blockNum). Uint64("myBlock", consensus.blockNum).
Uint64("myViewID", consensus.viewID). Uint64("myViewID", consensus.GetCurViewID()).
Interface("phase", consensus.phase). Interface("phase", consensus.phase).
Str("mode", consensus.current.Mode().String()). Str("mode", consensus.current.Mode().String()).
Logger() Logger()
@ -470,12 +453,12 @@ func (consensus *Consensus) UpdateConsensusInformation() Mode {
if len(curHeader.ShardState()) == 0 && curHeader.Number().Uint64() != 0 { if len(curHeader.ShardState()) == 0 && curHeader.Number().Uint64() != 0 {
leaderPubKey, err := consensus.getLeaderPubKeyFromCoinbase(curHeader) leaderPubKey, err := consensus.getLeaderPubKeyFromCoinbase(curHeader)
if err != nil || leaderPubKey == nil { if err != nil || leaderPubKey == nil {
consensus.getLogger().Debug().Err(err). consensus.getLogger().Error().Err(err).
Msg("[UpdateConsensusInformation] Unable to get leaderPubKey from coinbase") Msg("[UpdateConsensusInformation] Unable to get leaderPubKey from coinbase")
consensus.IgnoreViewIDCheck.Set() consensus.IgnoreViewIDCheck.Set()
hasError = true hasError = true
} else { } else {
consensus.getLogger().Debug(). consensus.getLogger().Info().
Str("leaderPubKey", leaderPubKey.Bytes.Hex()). Str("leaderPubKey", leaderPubKey.Bytes.Hex()).
Msg("[UpdateConsensusInformation] Most Recent LeaderPubKey Updated Based on BlockChain") Msg("[UpdateConsensusInformation] Most Recent LeaderPubKey Updated Based on BlockChain")
consensus.LeaderPubKey = leaderPubKey consensus.LeaderPubKey = leaderPubKey
@ -494,10 +477,8 @@ func (consensus *Consensus) UpdateConsensusInformation() Mode {
if (oldLeader != nil && consensus.LeaderPubKey != nil && if (oldLeader != nil && consensus.LeaderPubKey != nil &&
!consensus.LeaderPubKey.Object.IsEqual(oldLeader.Object)) && consensus.IsLeader() { !consensus.LeaderPubKey.Object.IsEqual(oldLeader.Object)) && consensus.IsLeader() {
go func() { go func() {
utils.Logger().Debug(). consensus.getLogger().Info().
Str("myKey", myPubKeys.SerializeToHexStr()). Str("myKey", myPubKeys.SerializeToHexStr()).
Uint64("viewID", consensus.viewID).
Uint64("block", consensus.blockNum).
Msg("[UpdateConsensusInformation] I am the New Leader") Msg("[UpdateConsensusInformation] I am the New Leader")
consensus.ReadySignal <- struct{}{} consensus.ReadySignal <- struct{}{}
}() }()
@ -553,3 +534,20 @@ func (consensus *Consensus) addViewIDKeyIfNotExist(viewID uint64) {
consensus.viewIDBitmap[viewID] = viewIDBitmap consensus.viewIDBitmap[viewID] = viewIDBitmap
} }
} }
// SetViewID 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) SetViewID(height uint64) {
consensus.SetCurViewID(height)
consensus.SetViewChangingID(height)
}
// SetCurViewID set the current view ID
func (consensus *Consensus) SetCurViewID(viewID uint64) {
consensus.current.SetCurViewID(viewID)
}
// SetViewChangingID set the current view change ID
func (consensus *Consensus) SetViewChangingID(viewID uint64) {
consensus.current.SetViewChangingID(viewID)
}

41
consensus/consensus_v2.go
Unescape View File

@ -163,9 +163,9 @@ func (consensus *Consensus) finalizeCommits() {
if consensus.consensusTimeout[timeoutBootstrap].IsActive() { if consensus.consensusTimeout[timeoutBootstrap].IsActive() {
consensus.consensusTimeout[timeoutBootstrap].Stop() consensus.consensusTimeout[timeoutBootstrap].Stop()
consensus.getLogger().Debug().Msg("[finalizeCommits] Start consensus timer; stop bootstrap timer only once") consensus.getLogger().Info().Msg("[finalizeCommits] Start consensus timer; stop bootstrap timer only once")
} else { } else {
consensus.getLogger().Debug().Msg("[finalizeCommits] Start consensus timer") consensus.getLogger().Info().Msg("[finalizeCommits] Start consensus timer")
} }
consensus.consensusTimeout[timeoutConsensus].Start() consensus.consensusTimeout[timeoutConsensus].Start()
@ -179,7 +179,7 @@ func (consensus *Consensus) finalizeCommits() {
Msg("HOORAY!!!!!!! CONSENSUS REACHED!!!!!!!") Msg("HOORAY!!!!!!! CONSENSUS REACHED!!!!!!!")
// Sleep to wait for the full block time // Sleep to wait for the full block time
consensus.getLogger().Debug().Msg("[finalizeCommits] Waiting for Block Time") consensus.getLogger().Info().Msg("[finalizeCommits] Waiting for Block Time")
<-time.After(time.Until(consensus.NextBlockDue)) <-time.After(time.Until(consensus.NextBlockDue))
// Send signal to Node to propose the new block for consensus // Send signal to Node to propose the new block for consensus
@ -281,7 +281,7 @@ func (consensus *Consensus) tryCatchup() {
consensus.getLogger().Info().Msg("[TryCatchup] block found to commit") consensus.getLogger().Info().Msg("[TryCatchup] block found to commit")
atomic.AddUint64(&consensus.blockNum, 1) atomic.AddUint64(&consensus.blockNum, 1)
atomic.StoreUint64(&consensus.viewID, committedMsg.ViewID+1) consensus.SetCurViewID(committedMsg.ViewID + 1)
consensus.LeaderPubKey = committedMsg.SenderPubkeys[0] consensus.LeaderPubKey = committedMsg.SenderPubkeys[0]
@ -344,10 +344,7 @@ func (consensus *Consensus) Start(
ticker := time.NewTicker(3 * time.Second) ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop() defer ticker.Stop()
consensus.consensusTimeout[timeoutBootstrap].Start() consensus.consensusTimeout[timeoutBootstrap].Start()
consensus.getLogger().Debug(). consensus.getLogger().Info().Msg("[ConsensusMainLoop] Start bootstrap timeout (only once)")
Uint64("viewID", consensus.viewID).
Uint64("blockNum", consensus.blockNum).
Msg("[ConsensusMainLoop] Start bootstrap timeout (only once)")
vdfInProgress := false vdfInProgress := false
// Set up next block due time. // Set up next block due time.
@ -358,7 +355,7 @@ func (consensus *Consensus) Start(
case <-toStart: case <-toStart:
start = true start = true
case <-ticker.C: case <-ticker.C:
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] Ticker") consensus.getLogger().Info().Msg("[ConsensusMainLoop] Ticker")
if !start && isInitialLeader { if !start && isInitialLeader {
continue continue
} }
@ -371,26 +368,28 @@ func (consensus *Consensus) Start(
continue continue
} }
if k != timeoutViewChange { if k != timeoutViewChange {
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] Ops Consensus Timeout!!!") consensus.getLogger().Warn().Msg("[ConsensusMainLoop] Ops Consensus Timeout!!!")
consensus.startViewChange(consensus.viewID + 1) viewID := consensus.GetCurViewID()
consensus.startViewChange(viewID + 1)
break break
} else { } else {
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] Ops View Change Timeout!!!") consensus.getLogger().Warn().Msg("[ConsensusMainLoop] Ops View Change Timeout!!!")
viewID := consensus.current.ViewID() viewID := consensus.GetViewChangingID()
consensus.startViewChange(viewID + 1) consensus.startViewChange(viewID + 1)
break break
} }
} }
case <-consensus.syncReadyChan: case <-consensus.syncReadyChan:
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] syncReadyChan") consensus.getLogger().Info().Msg("[ConsensusMainLoop] syncReadyChan")
consensus.SetBlockNum(consensus.ChainReader.CurrentHeader().Number().Uint64() + 1) consensus.SetBlockNum(consensus.ChainReader.CurrentHeader().Number().Uint64() + 1)
consensus.SetViewID(consensus.ChainReader.CurrentHeader().ViewID().Uint64() + 1) consensus.SetCurViewID(consensus.ChainReader.CurrentHeader().ViewID().Uint64() + 1)
consensus.SetViewChangingID(consensus.ChainReader.CurrentHeader().ViewID().Uint64() + 1)
mode := consensus.UpdateConsensusInformation() mode := consensus.UpdateConsensusInformation()
consensus.current.SetMode(mode) consensus.current.SetMode(mode)
consensus.getLogger().Info().Str("Mode", mode.String()).Msg("Node is IN SYNC") consensus.getLogger().Info().Str("Mode", mode.String()).Msg("Node is IN SYNC")
case <-consensus.syncNotReadyChan: case <-consensus.syncNotReadyChan:
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] syncNotReadyChan") consensus.getLogger().Info().Msg("[ConsensusMainLoop] syncNotReadyChan")
consensus.SetBlockNum(consensus.ChainReader.CurrentHeader().Number().Uint64() + 1) consensus.SetBlockNum(consensus.ChainReader.CurrentHeader().Number().Uint64() + 1)
consensus.current.SetMode(Syncing) consensus.current.SetMode(Syncing)
consensus.getLogger().Info().Msg("[ConsensusMainLoop] Node is OUT OF SYNC") consensus.getLogger().Info().Msg("[ConsensusMainLoop] Node is OUT OF SYNC")
@ -476,7 +475,7 @@ func (consensus *Consensus) Start(
startTime = time.Now() startTime = time.Now()
consensus.msgSender.Reset(newBlock.NumberU64()) consensus.msgSender.Reset(newBlock.NumberU64())
consensus.getLogger().Debug(). consensus.getLogger().Info().
Int("numTxs", len(newBlock.Transactions())). Int("numTxs", len(newBlock.Transactions())).
Int("numStakingTxs", len(newBlock.StakingTransactions())). Int("numStakingTxs", len(newBlock.StakingTransactions())).
Time("startTime", startTime). Time("startTime", startTime).
@ -485,23 +484,23 @@ func (consensus *Consensus) Start(
consensus.announce(newBlock) consensus.announce(newBlock)
case viewID := <-consensus.commitFinishChan: case viewID := <-consensus.commitFinishChan:
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] commitFinishChan") consensus.getLogger().Info().Msg("[ConsensusMainLoop] commitFinishChan")
// Only Leader execute this condition // Only Leader execute this condition
func() { func() {
consensus.mutex.Lock() consensus.mutex.Lock()
defer consensus.mutex.Unlock() defer consensus.mutex.Unlock()
if viewID == consensus.viewID { if viewID == consensus.GetCurViewID() {
consensus.finalizeCommits() consensus.finalizeCommits()
} }
}() }()
case <-stopChan: case <-stopChan:
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] stopChan") consensus.getLogger().Info().Msg("[ConsensusMainLoop] stopChan")
return return
} }
} }
consensus.getLogger().Debug().Msg("[ConsensusMainLoop] Ended.") consensus.getLogger().Info().Msg("[ConsensusMainLoop] Ended.")
}() }()
} }

6
consensus/consensus_viewchange_msg.go
Unescape View File

@ -24,7 +24,7 @@ func (consensus *Consensus) constructViewChangeMessage(priKey *bls.PrivateKeyWra
} }
vcMsg := message.GetViewchange() vcMsg := message.GetViewchange()
vcMsg.ViewId = consensus.current.ViewID() vcMsg.ViewId = consensus.GetViewChangingID()
vcMsg.BlockNum = consensus.blockNum vcMsg.BlockNum = consensus.blockNum
vcMsg.ShardId = consensus.ShardID vcMsg.ShardId = consensus.ShardID
// sender address // sender address
@ -82,7 +82,7 @@ func (consensus *Consensus) constructViewChangeMessage(priKey *bls.PrivateKeyWra
} }
viewIDBytes := make([]byte, 8) viewIDBytes := make([]byte, 8)
binary.LittleEndian.PutUint64(viewIDBytes, consensus.current.ViewID()) binary.LittleEndian.PutUint64(viewIDBytes, consensus.GetViewChangingID())
sign1 := priKey.Pri.SignHash(viewIDBytes) sign1 := priKey.Pri.SignHash(viewIDBytes)
if sign1 != nil { if sign1 != nil {
vcMsg.ViewidSig = sign1.Serialize() vcMsg.ViewidSig = sign1.Serialize()
@ -109,7 +109,7 @@ func (consensus *Consensus) constructNewViewMessage(viewID uint64, priKey *bls.P
} }
vcMsg := message.GetViewchange() vcMsg := message.GetViewchange()
vcMsg.ViewId = consensus.current.ViewID() vcMsg.ViewId = consensus.GetViewChangingID()
vcMsg.BlockNum = consensus.blockNum vcMsg.BlockNum = consensus.blockNum
vcMsg.ShardId = consensus.ShardID vcMsg.ShardId = consensus.ShardID
// sender address // sender address

2
consensus/construct.go
Unescape View File

@ -29,7 +29,7 @@ type NetworkMessage struct {
func (consensus *Consensus) populateMessageFields( func (consensus *Consensus) populateMessageFields(
request *msg_pb.ConsensusRequest, blockHash []byte, request *msg_pb.ConsensusRequest, blockHash []byte,
) *msg_pb.ConsensusRequest { ) *msg_pb.ConsensusRequest {
request.ViewId = consensus.viewID request.ViewId = consensus.GetCurViewID()
request.BlockNum = consensus.blockNum request.BlockNum = consensus.blockNum
request.ShardId = consensus.ShardID request.ShardId = consensus.ShardID
// 32 byte block hash // 32 byte block hash

12
consensus/debug.go
Unescape View File

@ -10,12 +10,12 @@ func (c *Consensus) GetConsensusMode() string {
return c.current.mode.String() return c.current.mode.String()
} }
// GetConsensusCurViewID .. // GetCurViewID ..
func (c *Consensus) GetConsensusCurViewID() uint64 { func (c *Consensus) GetCurViewID() uint64 {
return c.current.GetViewID() return c.current.GetCurViewID()
} }
// GetConsensusViewID .. // GetViewID ..
func (c *Consensus) GetConsensusViewID() uint64 { func (c *Consensus) GetViewChangingID() uint64 {
return c.viewID return c.current.GetViewChangingID()
} }

5
consensus/leader.go
Unescape View File

@ -110,12 +110,11 @@ func (consensus *Consensus) onPrepare(msg *msg_pb.Message) {
// TODO(audit): make FBFT lookup using map instead of looping through all items. // TODO(audit): make FBFT lookup using map instead of looping through all items.
if !consensus.FBFTLog.HasMatchingViewAnnounce( if !consensus.FBFTLog.HasMatchingViewAnnounce(
consensus.blockNum, consensus.viewID, recvMsg.BlockHash, consensus.blockNum, consensus.GetCurViewID(), recvMsg.BlockHash,
) { ) {
consensus.getLogger().Debug(). consensus.getLogger().Debug().
Uint64("MsgViewID", recvMsg.ViewID). Uint64("MsgViewID", recvMsg.ViewID).
Uint64("MsgBlockNum", recvMsg.BlockNum). Uint64("MsgBlockNum", recvMsg.BlockNum).
Uint64("blockNum", consensus.blockNum).
Msg("[OnPrepare] No Matching Announce message") Msg("[OnPrepare] No Matching Announce message")
} }
@ -288,7 +287,7 @@ func (consensus *Consensus) onCommit(msg *msg_pb.Message) {
//// Write - End //// Write - End
//// Read - Start //// Read - Start
viewID := consensus.viewID viewID := consensus.GetCurViewID()
if consensus.Decider.IsAllSigsCollected() { if consensus.Decider.IsAllSigsCollected() {
go func(viewID uint64) { go func(viewID uint64) {

97
consensus/view_change.go
Unescape View File

@ -28,38 +28,74 @@ const MaxViewIDDiff = 100
// State contains current mode and current viewID // State contains current mode and current viewID
type State struct { type State struct {
mode Mode mode Mode
viewID uint64 modeMux sync.RWMutex
mux sync.Mutex
// current view id in normal mode
// it changes per successful consensus
curViewID uint64
cViewMux sync.RWMutex
// view changing id is used during view change mode
// it is the next view id
viewChangingID uint64
viewMux sync.RWMutex
} }
// Mode return the current node mode // Mode return the current node mode
func (pm *State) Mode() Mode { func (pm *State) Mode() Mode {
pm.modeMux.RLock()
defer pm.modeMux.RUnlock()
return pm.mode return pm.mode
} }
// SetMode set the node mode as required // SetMode set the node mode as required
func (pm *State) SetMode(s Mode) { func (pm *State) SetMode(s Mode) {
pm.mux.Lock() pm.modeMux.Lock()
defer pm.mux.Unlock() defer pm.modeMux.Unlock()
pm.mode = s pm.mode = s
} }
// ViewID return the current viewchanging id // GetCurViewID return the current view id
func (pm *State) ViewID() uint64 { func (pm *State) GetCurViewID() uint64 {
return pm.viewID pm.cViewMux.RLock()
defer pm.cViewMux.RUnlock()
return pm.curViewID
}
// SetCurViewID sets the current view id
func (pm *State) SetCurViewID(viewID uint64) {
pm.cViewMux.Lock()
defer pm.cViewMux.Unlock()
pm.curViewID = viewID
} }
// SetViewID sets the viewchanging id accordingly // GetViewChangingID return the current view changing id
func (pm *State) SetViewID(viewID uint64) { // It is meaningful during view change mode
pm.mux.Lock() func (pm *State) GetViewChangingID() uint64 {
defer pm.mux.Unlock() pm.viewMux.RLock()
pm.viewID = viewID defer pm.viewMux.RUnlock()
return pm.viewChangingID
} }
// GetViewID returns the current viewchange viewID // SetViewChangingID set the current view changing id
func (pm *State) GetViewID() uint64 { // It is meaningful during view change mode
return pm.viewID func (pm *State) SetViewChangingID(id uint64) {
pm.viewMux.Lock()
defer pm.viewMux.Unlock()
pm.viewChangingID = id
}
// GetViewChangeDuraion return the duration of the current view change
// It increase in the power of difference betweeen view changing ID and current view ID
func (pm *State) GetViewChangeDuraion() time.Duration {
pm.viewMux.RLock()
pm.cViewMux.RLock()
defer pm.viewMux.RUnlock()
defer pm.cViewMux.RUnlock()
diff := int64(pm.viewChangingID - pm.curViewID)
return time.Duration(diff * diff * int64(viewChangeDuration))
} }
// switchPhase will switch FBFTPhase to nextPhase if the desirePhase equals the nextPhase // switchPhase will switch FBFTPhase to nextPhase if the desirePhase equals the nextPhase
@ -126,13 +162,12 @@ func (consensus *Consensus) startViewChange(viewID uint64) {
consensus.consensusTimeout[timeoutConsensus].Stop() consensus.consensusTimeout[timeoutConsensus].Stop()
consensus.consensusTimeout[timeoutBootstrap].Stop() consensus.consensusTimeout[timeoutBootstrap].Stop()
consensus.current.SetMode(ViewChanging) consensus.current.SetMode(ViewChanging)
consensus.current.SetViewID(viewID) consensus.SetViewChangingID(viewID)
consensus.LeaderPubKey = consensus.GetNextLeaderKey() consensus.LeaderPubKey = consensus.GetNextLeaderKey()
diff := int64(viewID - consensus.viewID) duration := consensus.current.GetViewChangeDuraion()
duration := time.Duration(diff * diff * int64(viewChangeDuration)) consensus.getLogger().Warn().
consensus.getLogger().Info(). Uint64("viewChangingID", consensus.GetViewChangingID()).
Uint64("ViewChangingID", viewID).
Dur("timeoutDuration", duration). Dur("timeoutDuration", duration).
Str("NextLeader", consensus.LeaderPubKey.Bytes.Hex()). Str("NextLeader", consensus.LeaderPubKey.Bytes.Hex()).
Msg("[startViewChange]") Msg("[startViewChange]")
@ -152,7 +187,7 @@ func (consensus *Consensus) startViewChange(viewID uint64) {
consensus.consensusTimeout[timeoutViewChange].SetDuration(duration) consensus.consensusTimeout[timeoutViewChange].SetDuration(duration)
consensus.consensusTimeout[timeoutViewChange].Start() consensus.consensusTimeout[timeoutViewChange].Start()
consensus.getLogger().Info(). consensus.getLogger().Info().
Uint64("ViewChangingID", consensus.current.ViewID()). Uint64("viewChangingID", consensus.GetViewChangingID()).
Msg("[startViewChange] start view change timer") Msg("[startViewChange] start view change timer")
} }
@ -396,7 +431,7 @@ func (consensus *Consensus) onViewChange(msg *msg_pb.Message) {
if len(consensus.m1Payload) == 0 { if len(consensus.m1Payload) == 0 {
// TODO(Chao): explain why ReadySignal is sent only in this case but not the other case. // TODO(Chao): explain why ReadySignal is sent only in this case but not the other case.
// Make sure the newly proposed block have the correct view ID // Make sure the newly proposed block have the correct view ID
consensus.viewID = recvMsg.ViewID consensus.SetCurViewID(recvMsg.ViewID)
go func() { go func() {
consensus.ReadySignal <- struct{}{} consensus.ReadySignal <- struct{}{}
}() }()
@ -447,7 +482,7 @@ func (consensus *Consensus) onViewChange(msg *msg_pb.Message) {
} }
} }
consensus.current.SetViewID(recvMsg.ViewID) consensus.SetCurViewID(recvMsg.ViewID)
msgToSend := consensus.constructNewViewMessage( msgToSend := consensus.constructNewViewMessage(
recvMsg.ViewID, newLeaderPriKey, recvMsg.ViewID, newLeaderPriKey,
) )
@ -467,18 +502,11 @@ func (consensus *Consensus) onViewChange(msg *msg_pb.Message) {
Msg("could not send out the NEWVIEW message") Msg("could not send out the NEWVIEW message")
} }
consensus.viewID = recvMsg.ViewID consensus.SetCurViewID(recvMsg.ViewID)
consensus.ResetViewChangeState() consensus.ResetViewChangeState()
consensus.consensusTimeout[timeoutViewChange].Stop() consensus.consensusTimeout[timeoutViewChange].Stop()
consensus.consensusTimeout[timeoutConsensus].Start() consensus.consensusTimeout[timeoutConsensus].Start()
consensus.getLogger().Debug(). consensus.getLogger().Info().Str("myKey", newLeaderKey.Bytes.Hex()).Msg("[onViewChange] I am the New Leader")
Uint64("viewChangingID", consensus.current.ViewID()).
Msg("[onViewChange] New Leader Start Consensus Timer and Stop View Change Timer")
consensus.getLogger().Info().
Str("myKey", newLeaderKey.Bytes.Hex()).
Uint64("viewID", consensus.viewID).
Uint64("block", consensus.blockNum).
Msg("[onViewChange] I am the New Leader")
} }
} }
@ -612,8 +640,7 @@ func (consensus *Consensus) onNewView(msg *msg_pb.Message) {
} }
// newView message verified success, override my state // newView message verified success, override my state
consensus.viewID = recvMsg.ViewID consensus.SetCurViewID(recvMsg.ViewID)
consensus.current.SetViewID(recvMsg.ViewID)
consensus.LeaderPubKey = senderKey consensus.LeaderPubKey = senderKey
consensus.ResetViewChangeState() consensus.ResetViewChangeState()

4
consensus/view_change_test.go
Unescape View File

@ -27,13 +27,13 @@ func TestBasicViewChanging(t *testing.T) {
// Change ViewID // Change ViewID
assert.Equal(t, state.viewID, consensus.current.viewID) assert.Equal(t, state.viewID, consensus.current.viewID)
assert.Equal(t, state.ViewID(), consensus.current.ViewID()) assert.Equal(t, state.ViewID(), consensus.current.ViewID())
assert.Equal(t, state.GetViewID(), consensus.current.GetViewID()) // Why are there two methods to retrieve the ViewID? assert.Equal(t, state.GetViewID(), consensus.GetViewID()) // Why are there two methods to retrieve the ViewID?
newViewID := consensus.current.ViewID() + 1 newViewID := consensus.current.ViewID() + 1
consensus.current.SetViewID(newViewID) consensus.current.SetViewID(newViewID)
assert.Equal(t, newViewID, consensus.current.viewID) assert.Equal(t, newViewID, consensus.current.viewID)
assert.Equal(t, newViewID, consensus.current.ViewID()) assert.Equal(t, newViewID, consensus.current.ViewID())
assert.Equal(t, newViewID, consensus.current.GetViewID()) assert.Equal(t, newViewID, consensus.GetViewID())
} }
func TestPhaseSwitching(t *testing.T) { func TestPhaseSwitching(t *testing.T) {

2
hmy/hmy.go
Unescape View File

@ -96,7 +96,7 @@ type NodeAPI interface {
// debug API // debug API
GetConsensusMode() string GetConsensusMode() string
GetConsensusPhase() string GetConsensusPhase() string
GetConsensusViewID() uint64 GetConsensusViewChangingID() uint64
GetConsensusCurViewID() uint64 GetConsensusCurViewID() uint64
ShutDown() ShutDown()
} }

10
node/api.go
Unescape View File

@ -117,12 +117,12 @@ func (node *Node) GetConsensusPhase() string {
return node.Consensus.GetConsensusPhase() return node.Consensus.GetConsensusPhase()
} }
// GetConsensusViewID returns the consensus.viewID // GetConsensusViewChangingID returns the view changing ID
func (node *Node) GetConsensusViewID() uint64 { func (node *Node) GetConsensusViewChangingID() uint64 {
return node.Consensus.GetConsensusViewID() return node.Consensus.GetViewChangingID()
} }
// GetConsensusCurViewID returns the consensus.current.viewID // GetConsensusCurViewID returns the current view ID
func (node *Node) GetConsensusCurViewID() uint64 { func (node *Node) GetConsensusCurViewID() uint64 {
return node.Consensus.GetConsensusCurViewID() return node.Consensus.GetCurViewID()
} }

2
node/node_newblock.go
Unescape View File

@ -232,7 +232,7 @@ func (node *Node) proposeNewBlock() (*types.Block, error) {
} }
return node.Worker.FinalizeNewBlock( return node.Worker.FinalizeNewBlock(
sig, mask, node.Consensus.GetViewID(), sig, mask, node.Consensus.GetCurViewID(),
coinbase, crossLinksToPropose, shardState, coinbase, crossLinksToPropose, shardState,
) )
} }

4
rpc/debug.go
Unescape View File

@ -39,10 +39,10 @@ func (*PrivateDebugService) SetLogVerbosity(ctx context.Context, level int) (map
return map[string]interface{}{"verbosity": verbosity.String()}, nil return map[string]interface{}{"verbosity": verbosity.String()}, nil
} }
func (s *PrivateDebugService) ConsensusViewID( func (s *PrivateDebugService) ConsensusViewChangingID(
ctx context.Context, ctx context.Context,
) uint64 { ) uint64 {
return s.hmy.NodeAPI.GetConsensusViewID() return s.hmy.NodeAPI.GetConsensusViewChangingID()
} }
func (s *PrivateDebugService) ConsensusCurViewID( func (s *PrivateDebugService) ConsensusCurViewID(

Write Preview
Loading…
Cancel
Save