diff options
Diffstat (limited to 'vendor/github.com/dexon-foundation/dexon-consensus/core/syncer')
-rw-r--r-- | vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/agreement.go | 18 | ||||
-rw-r--r-- | vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/consensus.go | 88 |
2 files changed, 31 insertions, 75 deletions
diff --git a/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/agreement.go b/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/agreement.go index 9f1abcaf5..f172b3b4c 100644 --- a/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/agreement.go +++ b/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/agreement.go @@ -103,7 +103,7 @@ func (a *agreement) processBlock(b *types.Block) { func (a *agreement) processAgreementResult(r *types.AgreementResult) { // Cache those results that CRS is not ready yet. if _, exists := a.confirmedBlocks[r.BlockHash]; exists { - a.logger.Trace("agreement result already confirmed", "result", r) + a.logger.Trace("Agreement result already confirmed", "result", r) return } if r.Position.Round > a.latestCRSRound { @@ -113,11 +113,11 @@ func (a *agreement) processAgreementResult(r *types.AgreementResult) { a.pendings[r.Position.Round] = pendingsForRound } pendingsForRound[r.BlockHash] = r - a.logger.Trace("agreement result cached", "result", r) + a.logger.Trace("Agreement result cached", "result", r) return } if err := core.VerifyAgreementResult(r, a.cache); err != nil { - a.logger.Error("agreement result verification failed", + a.logger.Error("Agreement result verification failed", "result", r, "error", err) return @@ -144,12 +144,12 @@ loop: case a.pullChan <- r.BlockHash: break loop case <-a.ctx.Done(): - a.logger.Error("pull request is not sent", + a.logger.Error("Pull request is not sent", "position", &r.Position, "hash", r.BlockHash.String()[:6]) return case <-time.After(500 * time.Millisecond): - a.logger.Debug("pull request is unable to send", + a.logger.Debug("Pull request is unable to send", "position", &r.Position, "hash", r.BlockHash.String()[:6]) } @@ -171,12 +171,12 @@ func (a *agreement) processNewCRS(round uint64) { delete(a.pendings, r) for _, res := range pendingsForRound { if err := core.VerifyAgreementResult(res, a.cache); err != nil { - a.logger.Error("invalid agreement result", + a.logger.Error("Invalid agreement result", "result", res, "error", err) continue } - a.logger.Error("flush agreement result", "result", res) + a.logger.Error("Flush agreement result", "result", res) a.processAgreementResult(res) break } @@ -194,10 +194,10 @@ func (a *agreement) confirm(b *types.Block) { case a.outputChan <- b: break loop case <-a.ctx.Done(): - a.logger.Error("confirmed block is not sent", "block", b) + a.logger.Error("Confirmed block is not sent", "block", b) return case <-time.After(500 * time.Millisecond): - a.logger.Debug("agreement output channel is full", "block", b) + a.logger.Debug("Agreement output channel is full", "block", b) } } a.confirmedBlocks[b.Hash] = struct{}{} diff --git a/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/consensus.go b/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/consensus.go index 4fc24b407..24c781aac 100644 --- a/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/consensus.go +++ b/vendor/github.com/dexon-foundation/dexon-consensus/core/syncer/consensus.go @@ -66,8 +66,6 @@ type Consensus struct { randomnessResults map[common.Hash]*types.BlockRandomnessResult blocks types.BlocksByPosition agreementModule *agreement - configs []*types.Config - roundBeginHeights []uint64 agreementRoundCut uint64 heightEvt *common.Event roundEvt *utils.RoundEvent @@ -102,19 +100,15 @@ func NewConsensus( logger common.Logger) *Consensus { con := &Consensus{ - dMoment: dMoment, - app: app, - gov: gov, - db: db, - network: network, - nodeSetCache: utils.NewNodeSetCache(gov), - tsigVerifier: core.NewTSigVerifierCache(gov, 7), - prv: prv, - logger: logger, - configs: []*types.Config{ - utils.GetConfigWithPanic(gov, 0, logger), - }, - roundBeginHeights: []uint64{0}, + dMoment: dMoment, + app: app, + gov: gov, + db: db, + network: network, + nodeSetCache: utils.NewNodeSetCache(gov), + tsigVerifier: core.NewTSigVerifierCache(gov, 7), + prv: prv, + logger: logger, receiveChan: make(chan *types.Block, 1000), pullChan: make(chan common.Hash, 1000), randomnessResults: make(map[common.Hash]*types.BlockRandomnessResult), @@ -153,13 +147,12 @@ func (con *Consensus) assureBuffering() { ) if height == 0 { con.roundEvt, err = utils.NewRoundEvent(con.ctx, con.gov, con.logger, - uint64(0), uint64(0), uint64(0), core.ConfigRoundShift) + 0, 0, core.ConfigRoundShift) } else { var b types.Block if b, err = con.db.GetBlock(blockHash); err == nil { - beginHeight := con.roundBeginHeights[b.Position.Round] con.roundEvt, err = utils.NewRoundEvent(con.ctx, con.gov, - con.logger, b.Position.Round, beginHeight, beginHeight, + con.logger, b.Position.Round, b.Finalization.Height, core.ConfigRoundShift) } } @@ -176,6 +169,7 @@ func (con *Consensus) assureBuffering() { continue } con.nodeSetCache.Purge(e.Round + 1) + con.tsigVerifier.Purge(e.Round + 1) } }) // Register a round event handler to notify CRS to agreementModule. @@ -197,7 +191,7 @@ func (con *Consensus) assureBuffering() { return false case <-time.After(500 * time.Millisecond): con.logger.Warn( - "agreement input channel is full when putting CRS", + "Agreement input channel is full when notifying new round", "round", e.Round, ) return true @@ -223,7 +217,7 @@ func (con *Consensus) checkIfSynced(blocks []*types.Block) (synced bool) { con.lock.RLock() defer con.lock.RUnlock() defer func() { - con.logger.Debug("syncer synced status", + con.logger.Debug("Syncer synced status", "last-block", blocks[len(blocks)-1], "synced", synced, ) @@ -264,7 +258,6 @@ func (con *Consensus) ForceSync(skip bool) { panic(err) } con.logger.Info("Force Sync", "block", &block) - con.setupConfigsUntilRound(block.Position.Round + core.ConfigRoundShift - 1) con.syncedLastBlock = &block con.stopBuffering() // We might call stopBuffering without calling assureBuffering. @@ -310,20 +303,19 @@ func (con *Consensus) SyncBlocks( // tip in DB. _, tipHeight := con.db.GetCompactionChainTipInfo() if blocks[0].Finalization.Height != tipHeight+1 { - con.logger.Error("mismatched finalization height", + con.logger.Error("Mismatched finalization height", "now", blocks[0].Finalization.Height, "expected", tipHeight+1, ) err = ErrInvalidSyncingFinalizationHeight return } - con.logger.Trace("syncBlocks", + con.logger.Trace("SyncBlocks", "position", &blocks[0].Position, "final height", blocks[0].Finalization.Height, "len", len(blocks), "latest", latest, ) - con.setupConfigs(blocks) for _, b := range blocks { if err = con.db.PutBlock(*b); err != nil { // A block might be put into db when confirmed by BA, but not @@ -339,7 +331,7 @@ func (con *Consensus) SyncBlocks( b.Hash, b.Finalization.Height); err != nil { return } - go con.heightEvt.NotifyHeight(b.Finalization.Height) + con.heightEvt.NotifyHeight(b.Finalization.Height) } if latest { con.assureBuffering() @@ -377,7 +369,6 @@ func (con *Consensus) GetSyncedConsensus() (*core.Consensus, error) { var err error con.syncedConsensus, err = core.NewConsensusFromSyncer( con.syncedLastBlock, - con.roundBeginHeights[con.syncedLastBlock.Position.Round], con.syncedSkipNext, con.dMoment, con.app, @@ -413,14 +404,14 @@ func (con *Consensus) stopBuffering() { return } con.duringBuffering = false - con.logger.Trace("syncer is about to stop") + con.logger.Trace("Syncer is about to stop") // Stop network and CRS routines, wait until they are all stoped. con.ctxCancel() return }() { return } - con.logger.Trace("stop syncer modules") + con.logger.Trace("Stop syncer modules") con.roundEvt.Stop() con.waitGroup.Done() // Wait for all routines depends on con.agreementModule stopped. @@ -433,9 +424,9 @@ func (con *Consensus) stopBuffering() { con.dummyMsgBuffer = append(con.dummyMsgBuffer, msg) }) // Stop agreements. - con.logger.Trace("stop syncer agreement modules") + con.logger.Trace("Stop syncer agreement modules") con.stopAgreement() - con.logger.Trace("syncer stopped") + con.logger.Trace("Syncer stopped") return } @@ -447,48 +438,13 @@ func (con *Consensus) isEmptyBlock(b *types.Block) bool { // buildEmptyBlock builds an empty block in agreement. func (con *Consensus) buildEmptyBlock(b *types.Block, parent *types.Block) { - cfg := con.configs[b.Position.Round] + cfg := utils.GetConfigWithPanic(con.gov, b.Position.Round, con.logger) b.Timestamp = parent.Timestamp.Add(cfg.MinBlockInterval) b.Witness.Height = parent.Witness.Height b.Witness.Data = make([]byte, len(parent.Witness.Data)) copy(b.Witness.Data, parent.Witness.Data) } -// setupConfigs is called by SyncBlocks with blocks from compaction chain. In -// the first time, setupConfigs setups from round 0. -func (con *Consensus) setupConfigs(blocks []*types.Block) { - // Find max round in blocks. - var maxRound uint64 - for _, b := range blocks { - if b.Position.Round > maxRound { - maxRound = b.Position.Round - } - } - // Get configs from governance. - // - // In fullnode, the notification of new round is yet another TX, which - // needs to be executed after corresponding block delivered. Thus, the - // configuration for 'maxRound + core.ConfigRoundShift' won't be ready when - // seeing this block. - con.setupConfigsUntilRound(maxRound + core.ConfigRoundShift - 1) -} - -func (con *Consensus) setupConfigsUntilRound(round uint64) { - con.lock.Lock() - defer con.lock.Unlock() - con.logger.Debug("syncer setupConfigs", - "until-round", round, - "length", len(con.configs), - ) - for r := uint64(len(con.configs)); r <= round; r++ { - cfg := utils.GetConfigWithPanic(con.gov, r, con.logger) - con.configs = append(con.configs, cfg) - con.roundBeginHeights = append( - con.roundBeginHeights, - con.roundBeginHeights[r-1]+con.configs[r-1].RoundLength) - } -} - // startAgreement starts agreements for receiving votes and agreements. func (con *Consensus) startAgreement() { // Start a routine for listening receive channel and pull block channel. |