dex: fix fast sync in BP mode (#291)

Start blockproposer only after fast sync is completed (received
ChainHead event), so watchcat is not falsely triggered.

1 files changed, 38 insertions, 34 deletions

diff --git a/dex/handler.go b/dex/handler.go
index 245b31807..84170e54b 100644
--- a/dex/handler.go
+++ b/dex/handler.go
@@ -136,8 +136,8 @@ type ProtocolManager struct {
 	chainHeadSub event.Subscription
 
 	// channels for dexon consensus core
-	receiveCh      chan interface{}
-	receiveEnabled int32
+	receiveCh          chan interface{}
+	receiveCoreMessage int32
 
 	srvr p2pServer
 
@@ -166,29 +166,25 @@ func NewProtocolManager(
 	tab := newNodeTable()
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
-		networkID:        networkID,
-		eventMux:         mux,
-		txpool:           txpool,
-		nodeTable:        tab,
-		gov:              gov,
-		blockchain:       blockchain,
-		cache:            newCache(5120, dexDB.NewDatabase(chaindb)),
-		nextPullVote:     &sync.Map{},
-		chainconfig:      config,
-		newPeerCh:        make(chan *peer),
-		noMorePeers:      make(chan struct{}),
-		txsyncCh:         make(chan *txsync),
-		recordsyncCh:     make(chan *recordsync),
-		quitSync:         make(chan struct{}),
-		receiveCh:        make(chan interface{}, 1024),
-		receiveEnabled:   0,
-		isBlockProposer:  isBlockProposer,
-		app:              app,
-		blockNumberGauge: metrics.GetOrRegisterGauge("dex/blocknumber", nil),
-	}
-
-	if isBlockProposer {
-		manager.receiveEnabled = 1
+		networkID:          networkID,
+		eventMux:           mux,
+		txpool:             txpool,
+		nodeTable:          tab,
+		gov:                gov,
+		blockchain:         blockchain,
+		cache:              newCache(5120, dexDB.NewDatabase(chaindb)),
+		nextPullVote:       &sync.Map{},
+		chainconfig:        config,
+		newPeerCh:          make(chan *peer),
+		noMorePeers:        make(chan struct{}),
+		txsyncCh:           make(chan *txsync),
+		recordsyncCh:       make(chan *recordsync),
+		quitSync:           make(chan struct{}),
+		receiveCh:          make(chan interface{}, 1024),
+		receiveCoreMessage: 0,
+		isBlockProposer:    isBlockProposer,
+		app:                app,
+		blockNumberGauge:   metrics.GetOrRegisterGauge("dex/blocknumber", nil),
 	}
 
 	// Figure out whether to allow fast sync or not
@@ -839,7 +835,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 	// Block proposer-only messages.
 
 	case msg.Code == CoreBlockMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		var blocks []*coreTypes.Block
@@ -851,7 +847,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			pm.receiveCh <- block
 		}
 	case msg.Code == VoteMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		var votes []*coreTypes.Vote
@@ -865,7 +861,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			pm.receiveCh <- vote
 		}
 	case msg.Code == AgreementMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		// DKG set is receiver
@@ -876,7 +872,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		p.MarkAgreement(rlpHash(agreement))
 		pm.receiveCh <- &agreement
 	case msg.Code == RandomnessMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		// Broadcast this to all peer
@@ -889,7 +885,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			pm.receiveCh <- randomness
 		}
 	case msg.Code == DKGPrivateShareMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		// Do not relay this msg
@@ -900,7 +896,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		p.MarkDKGPrivateShares(rlpHash(ps))
 		pm.receiveCh <- &ps
 	case msg.Code == DKGPartialSignatureMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		// broadcast in DKG set
@@ -910,7 +906,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		pm.receiveCh <- &psig
 	case msg.Code == PullBlocksMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		var hashes coreCommon.Hashes
@@ -921,7 +917,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		log.Debug("Push blocks", "blocks", blocks)
 		return p.SendCoreBlocks(blocks)
 	case msg.Code == PullVotesMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		next, ok := pm.nextPullVote.Load(p.ID())
@@ -940,7 +936,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		log.Debug("Push votes", "votes", votes)
 		return p.SendVotes(votes)
 	case msg.Code == PullRandomnessMsg:
-		if atomic.LoadInt32(&pm.receiveEnabled) == 0 {
+		if atomic.LoadInt32(&pm.receiveCoreMessage) == 0 {
 			break
 		}
 		var hashes coreCommon.Hashes
@@ -1287,6 +1283,14 @@ func (pm *ProtocolManager) recordBroadcastLoop() {
 	}
 }
 
+func (pm *ProtocolManager) SetReceiveCoreMessage(enabled bool) {
+	if enabled {
+		atomic.StoreInt32(&pm.receiveCoreMessage, 1)
+	} else {
+		atomic.StoreInt32(&pm.receiveCoreMessage, 0)
+	}
+}
+
 // a loop keep building and maintaining peers in notary set.
 // TODO: finish this
 func (pm *ProtocolManager) peerSetLoop() {