diff options
Diffstat (limited to 'dex/handler.go')
| -rw-r--r-- | dex/handler.go | 72 |
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() { |