aboutsummaryrefslogtreecommitdiffstats
path: root/vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go')
-rw-r--r--vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go332
1 files changed, 127 insertions, 205 deletions
diff --git a/vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go b/vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go
index f46501038..f4c0a372d 100644
--- a/vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go
+++ b/vendor/github.com/dexon-foundation/dexon-consensus/core/consensus.go
@@ -57,10 +57,9 @@ var (
// consensusBAReceiver implements agreementReceiver.
type consensusBAReceiver struct {
- // TODO(mission): consensus would be replaced by lattice and network.
+ // TODO(mission): consensus would be replaced by blockChain and network.
consensus *Consensus
agreementModule *agreement
- chainID uint32
changeNotaryTime time.Time
roundValue *atomic.Value
isNotary bool
@@ -96,9 +95,9 @@ func (recv *consensusBAReceiver) ProposeBlock() common.Hash {
if !recv.isNotary {
return common.Hash{}
}
- block := recv.consensus.proposeBlock(recv.chainID, recv.round())
- if block == nil {
- recv.consensus.logger.Error("unable to propose block")
+ block, err := recv.consensus.proposeBlock(recv.agreementModule.agreementID())
+ if err != nil || block == nil {
+ recv.consensus.logger.Error("unable to propose block", "error", err)
return types.NullBlockHash
}
go func() {
@@ -115,25 +114,29 @@ func (recv *consensusBAReceiver) ProposeBlock() common.Hash {
func (recv *consensusBAReceiver) ConfirmBlock(
hash common.Hash, votes map[types.NodeID]*types.Vote) {
- var block *types.Block
+ var (
+ block *types.Block
+ aID = recv.agreementModule.agreementID()
+ )
isEmptyBlockConfirmed := hash == common.Hash{}
if isEmptyBlockConfirmed {
- aID := recv.agreementModule.agreementID()
- recv.consensus.logger.Info("Empty block is confirmed",
- "position", &aID)
+ recv.consensus.logger.Info("Empty block is confirmed", "position", aID)
var err error
- block, err = recv.consensus.proposeEmptyBlock(recv.round(), recv.chainID)
+ block, err = recv.consensus.bcModule.addEmptyBlock(aID)
if err != nil {
- recv.consensus.logger.Error("Propose empty block failed", "error", err)
+ recv.consensus.logger.Error("Add position for empty failed",
+ "error", err)
return
}
+ if block == nil {
+ panic(fmt.Errorf("empty block should be proposed directly: %s", aID))
+ }
} else {
var exist bool
block, exist = recv.agreementModule.findBlockNoLock(hash)
if !exist {
recv.consensus.logger.Error("Unknown block confirmed",
- "hash", hash.String()[:6],
- "chainID", recv.chainID)
+ "hash", hash.String()[:6])
ch := make(chan *types.Block)
func() {
recv.consensus.lock.Lock()
@@ -155,8 +158,7 @@ func (recv *consensusBAReceiver) ConfirmBlock(
}
recv.consensus.logger.Info("Receive unknown block",
"hash", hash.String()[:6],
- "position", &block.Position,
- "chainID", recv.chainID)
+ "position", block.Position)
recv.agreementModule.addCandidateBlock(block)
recv.agreementModule.lock.Lock()
defer recv.agreementModule.lock.Unlock()
@@ -165,15 +167,14 @@ func (recv *consensusBAReceiver) ConfirmBlock(
return
}
}
- recv.consensus.ccModule.registerBlock(block)
if block.Position.Height != 0 &&
- !recv.consensus.lattice.Exist(block.ParentHash) {
+ !recv.consensus.bcModule.confirmed(block.Position.Height-1) {
go func(hash common.Hash) {
parentHash := hash
for {
recv.consensus.logger.Warn("Parent block not confirmed",
"parent-hash", parentHash.String()[:6],
- "cur-position", &block.Position)
+ "cur-position", block.Position)
ch := make(chan *types.Block)
if !func() bool {
recv.consensus.lock.Lock()
@@ -200,13 +201,12 @@ func (recv *consensusBAReceiver) ConfirmBlock(
}
recv.consensus.logger.Info("Receive parent block",
"parent-hash", block.ParentHash.String()[:6],
- "cur-position", &block.Position,
- "chainID", recv.chainID)
- recv.consensus.ccModule.registerBlock(block)
+ "cur-position", block.Position)
recv.consensus.processBlockChan <- block
parentHash = block.ParentHash
if block.Position.Height == 0 ||
- recv.consensus.lattice.Exist(parentHash) {
+ recv.consensus.bcModule.confirmed(
+ block.Position.Height-1) {
return
}
}
@@ -372,11 +372,6 @@ type Consensus struct {
dkgReady *sync.Cond
cfgModule *configurationChain
- // Dexon consensus v1's modules.
- lattice *Lattice
- ccModule *compactionChain
- toSyncer *totalOrderingSyncer
-
// Interfaces.
db db.Database
app Application
@@ -385,21 +380,21 @@ type Consensus struct {
network Network
// Misc.
- dMoment time.Time
- nodeSetCache *utils.NodeSetCache
- round uint64
- roundForNewConfig uint64
- lock sync.RWMutex
- ctx context.Context
- ctxCancel context.CancelFunc
- event *common.Event
- logger common.Logger
- nonFinalizedBlockDelivered bool
- resetRandomnessTicker chan struct{}
- resetDeliveryGuardTicker chan struct{}
- msgChan chan interface{}
- waitGroup sync.WaitGroup
- processBlockChan chan *types.Block
+ bcModule *blockChain
+ dMoment time.Time
+ nodeSetCache *utils.NodeSetCache
+ round uint64
+ roundForNewConfig uint64
+ lock sync.RWMutex
+ ctx context.Context
+ ctxCancel context.CancelFunc
+ event *common.Event
+ logger common.Logger
+ resetRandomnessTicker chan struct{}
+ resetDeliveryGuardTicker chan struct{}
+ msgChan chan interface{}
+ waitGroup sync.WaitGroup
+ processBlockChan chan *types.Block
// Context of Dummy receiver during switching from syncer.
dummyCancel context.CancelFunc
@@ -417,7 +412,7 @@ func NewConsensus(
prv crypto.PrivateKey,
logger common.Logger) *Consensus {
return newConsensusForRound(
- &types.Block{}, dMoment, app, gov, db, network, prv, logger, nil, true)
+ nil, dMoment, app, gov, db, network, prv, logger, true)
}
// NewConsensusForSimulation creates an instance of Consensus for simulation,
@@ -431,7 +426,7 @@ func NewConsensusForSimulation(
prv crypto.PrivateKey,
logger common.Logger) *Consensus {
return newConsensusForRound(
- &types.Block{}, dMoment, app, gov, db, network, prv, logger, nil, false)
+ nil, dMoment, app, gov, db, network, prv, logger, false)
}
// NewConsensusFromSyncer constructs an Consensus instance from information
@@ -451,14 +446,13 @@ func NewConsensusFromSyncer(
db db.Database,
networkModule Network,
prv crypto.PrivateKey,
- latticeModule *Lattice,
- confirmedBlocks [][]*types.Block,
+ confirmedBlocks []*types.Block,
randomnessResults []*types.BlockRandomnessResult,
cachedMessages []interface{},
logger common.Logger) (*Consensus, error) {
// Setup Consensus instance.
con := newConsensusForRound(initBlock, initRoundBeginTime, app, gov, db,
- networkModule, prv, logger, latticeModule, true)
+ networkModule, prv, logger, true)
// Launch a dummy receiver before we start receiving from network module.
con.dummyMsgBuffer = cachedMessages
con.dummyCancel, con.dummyFinished = utils.LaunchDummyReceiver(
@@ -467,29 +461,18 @@ func NewConsensusFromSyncer(
})
// Dump all BA-confirmed blocks to the consensus instance, make sure these
// added blocks forming a DAG.
- for {
- updated := false
- for idx, bs := range confirmedBlocks {
- for bIdx, b := range bs {
- // Only when its parent block is already added to lattice, we can
- // then add this block. If not, our pulling mechanism would stop at
- // the block we added, and lost its parent block forever.
- if !latticeModule.Exist(b.ParentHash) {
- logger.Debug("Skip discontinuous confirmed block",
- "from", b,
- "until", bs[len(bs)-1])
- confirmedBlocks[idx] = bs[bIdx:]
- break
- }
- con.ccModule.registerBlock(b)
- if err := con.processBlock(b); err != nil {
- return nil, err
- }
- }
- }
- if !updated {
+ refBlock := initBlock
+ for _, b := range confirmedBlocks {
+ // Only when its parent block is already added to lattice, we can
+ // then add this block. If not, our pulling mechanism would stop at
+ // the block we added, and lost its parent block forever.
+ if b.Position.Height != refBlock.Position.Height+1 {
break
}
+ if err := con.processBlock(b); err != nil {
+ return nil, err
+ }
+ refBlock = b
}
// Dump all randomness result to the consensus instance.
for _, r := range randomnessResults {
@@ -502,7 +485,7 @@ func NewConsensusFromSyncer(
return con, nil
}
-// newConsensus creates a Consensus instance.
+// newConsensusForRound creates a Consensus instance.
func newConsensusForRound(
initBlock *types.Block,
initRoundBeginTime time.Time,
@@ -512,9 +495,7 @@ func newConsensusForRound(
network Network,
prv crypto.PrivateKey,
logger common.Logger,
- latticeModule *Lattice,
usingNonBlocking bool) *Consensus {
-
// TODO(w): load latest blockHeight from DB, and use config at that height.
nodeSetCache := utils.NewNodeSetCache(gov)
// Setup signer module.
@@ -525,13 +506,11 @@ func newConsensusForRound(
debugApp = a
}
// Get configuration for bootstrap round.
- initRound := initBlock.Position.Round
- initConfig := utils.GetConfigWithPanic(gov, initRound, logger)
- // Init lattice.
- if latticeModule == nil {
- latticeModule = NewLattice(initRoundBeginTime, initRound, initConfig,
- signer, app, debugApp, db, logger)
+ initRound := uint64(0)
+ if initBlock != nil {
+ initRound = initBlock.Position.Round
}
+ initConfig := utils.GetConfigWithPanic(gov, initRound, logger)
// Init configuration chain.
ID := types.NewNodeID(prv.PublicKey())
recv := &consensusDKGReceiver{
@@ -548,11 +527,14 @@ func newConsensusForRound(
if usingNonBlocking {
appModule = newNonBlocking(app, debugApp)
}
+ bcConfig := blockChainConfig{}
+ bcConfig.fromConfig(initRound, initConfig)
+ bcConfig.setRoundBeginTime(initRoundBeginTime)
+ bcModule := newBlockChain(ID, initBlock, bcConfig, appModule,
+ NewTSigVerifierCache(gov, 7), signer, logger)
// Construct Consensus instance.
con := &Consensus{
ID: ID,
- ccModule: newCompactionChain(gov),
- lattice: latticeModule,
app: appModule,
debugApp: debugApp,
gov: gov,
@@ -561,6 +543,7 @@ func newConsensusForRound(
baConfirmedBlock: make(map[common.Hash]chan<- *types.Block),
dkgReady: sync.NewCond(&sync.Mutex{}),
cfgModule: cfgModule,
+ bcModule: bcModule,
dMoment: initRoundBeginTime,
nodeSetCache: nodeSetCache,
signer: signer,
@@ -581,52 +564,56 @@ func newConsensusForRound(
// prepare the Consensus instance to be ready for blocks after 'initBlock'.
// 'initBlock' could be either:
-// - an empty block
+// - nil
// - the last finalized block
-func (con *Consensus) prepare(initBlock *types.Block) error {
+func (con *Consensus) prepare(initBlock *types.Block) (err error) {
// The block past from full node should be delivered already or known by
// full node. We don't have to notify it.
- con.roundForNewConfig = initBlock.Position.Round + 1
- initRound := initBlock.Position.Round
+ initRound := uint64(0)
+ if initBlock != nil {
+ initRound = initBlock.Position.Round
+ }
+ con.roundForNewConfig = initRound + 1
initConfig := utils.GetConfigWithPanic(con.gov, initRound, con.logger)
// Setup context.
- con.ccModule.init(initBlock)
con.logger.Debug("Calling Governance.CRS", "round", initRound)
initCRS := con.gov.CRS(initRound)
if (initCRS == common.Hash{}) {
- return ErrCRSNotReady
+ err = ErrCRSNotReady
+ return
}
- if err := con.baMgr.appendConfig(initRound, initConfig, initCRS); err != nil {
- return err
+ if err = con.baMgr.appendConfig(initRound, initConfig, initCRS); err != nil {
+ return
}
- // Setup lattice module.
+ // Setup blockChain module.
initPlusOneCfg := utils.GetConfigWithPanic(con.gov, initRound+1, con.logger)
- if err := con.lattice.AppendConfig(initRound+1, initPlusOneCfg); err != nil {
- if err == ErrRoundNotIncreasing {
- err = nil
- } else {
+ if err = con.bcModule.appendConfig(initRound+1, initPlusOneCfg); err != nil {
+ return
+ }
+ if initRound == 0 {
+ dkgSet, err := con.nodeSetCache.GetDKGSet(initRound)
+ if err != nil {
return err
}
+ if _, exist := dkgSet[con.ID]; exist {
+ con.logger.Info("Selected as DKG set", "round", initRound)
+ go func() {
+ // Sleep until dMoment come.
+ time.Sleep(con.dMoment.Sub(time.Now().UTC()))
+ // Network is not stable upon starting. Wait some time to ensure first
+ // DKG would success. Three is a magic number.
+ time.Sleep(initConfig.MinBlockInterval * 3)
+ con.cfgModule.registerDKG(initRound, getDKGThreshold(initConfig))
+ con.event.RegisterTime(con.dMoment.Add(initConfig.RoundInterval/4),
+ func(time.Time) {
+ con.runDKG(initRound, initConfig)
+ })
+ }()
+ }
}
// Register events.
- dkgSet, err := con.nodeSetCache.GetDKGSet(initRound)
- if err != nil {
- return err
- }
- if _, exist := dkgSet[con.ID]; exist {
- con.logger.Info("Selected as DKG set", "round", initRound)
- go func() {
- // Sleep until dMoment come.
- time.Sleep(con.dMoment.Sub(time.Now().UTC()))
- con.cfgModule.registerDKG(initRound, getDKGThreshold(initConfig))
- con.event.RegisterTime(con.dMoment.Add(initConfig.RoundInterval/4),
- func(time.Time) {
- con.runDKG(initRound, initConfig)
- })
- }()
- }
con.initialRound(con.dMoment, initRound, initConfig)
- return nil
+ return
}
// Run starts running DEXON Consensus.
@@ -847,7 +834,7 @@ func (con *Consensus) initialRound(
})
}(round + 1)
})
- // Prepare lattice module for next round and next "initialRound" routine.
+ // Prepare blockChain module for next round and next "initialRound" routine.
con.event.RegisterTime(startTime.Add(config.RoundInterval),
func(time.Time) {
// Change round.
@@ -917,7 +904,7 @@ MessageLoop:
ch, e := con.baConfirmedBlock[val.Hash]
return ch, e
}(); exist {
- if err := con.lattice.SanityCheck(val, false); err != nil {
+ if err := con.bcModule.sanityCheck(val); err != nil {
if err == ErrRetrySanityCheckLater {
err = nil
} else {
@@ -965,7 +952,7 @@ MessageLoop:
if err := con.ProcessBlockRandomnessResult(val, true); err != nil {
con.logger.Error("Failed to process block randomness result",
"hash", val.BlockHash.String()[:6],
- "position", &val.Position,
+ "position", val.Position,
"error", err)
}
case *typesDKG.PrivateShare:
@@ -983,34 +970,6 @@ MessageLoop:
}
}
-func (con *Consensus) proposeBlock(chainID uint32, round uint64) *types.Block {
- block := &types.Block{
- Position: types.Position{
- ChainID: chainID,
- Round: round,
- },
- }
- if err := con.prepareBlock(block, time.Now().UTC()); err != nil {
- con.logger.Error("Failed to prepare block", "error", err)
- return nil
- }
- return block
-}
-
-func (con *Consensus) proposeEmptyBlock(
- round uint64, chainID uint32) (*types.Block, error) {
- block := &types.Block{
- Position: types.Position{
- Round: round,
- ChainID: chainID,
- },
- }
- if err := con.lattice.PrepareEmptyBlock(block); err != nil {
- return nil, err
- }
- return block, nil
-}
-
// ProcessVote is the entry point to submit ont vote to a Consensus instance.
func (con *Consensus) ProcessVote(vote *types.Vote) (err error) {
v := vote.Clone()
@@ -1024,14 +983,11 @@ func (con *Consensus) ProcessAgreementResult(
if !con.baMgr.touchAgreementResult(rand) {
return nil
}
-
// Sanity Check.
if err := VerifyAgreementResult(rand, con.nodeSetCache); err != nil {
con.baMgr.untouchAgreementResult(rand)
return err
}
- con.lattice.AddShallowBlock(rand.BlockHash, rand.Position)
-
// Syncing BA Module.
if err := con.baMgr.processAgreementResult(rand); err != nil {
return err
@@ -1089,7 +1045,7 @@ func (con *Consensus) ProcessAgreementResult(
if err != nil {
if err != ErrTSigAlreadyRunning {
con.logger.Error("Failed to run TSIG",
- "position", &rand.Position,
+ "position", rand.Position,
"hash", rand.BlockHash.String()[:6],
"error", err)
}
@@ -1113,15 +1069,8 @@ func (con *Consensus) ProcessBlockRandomnessResult(
if rand.Position.Round == 0 {
return nil
}
- if !con.ccModule.touchBlockRandomnessResult(rand) {
- return nil
- }
- if err := con.ccModule.processBlockRandomnessResult(rand); err != nil {
- if err == ErrBlockNotRegistered {
- err = nil
- } else {
- return err
- }
+ if err := con.bcModule.addRandomness(rand); err != nil {
+ return err
}
if needBroadcast {
con.logger.Debug("Calling Network.BroadcastRandomnessResult",
@@ -1154,7 +1103,7 @@ func (con *Consensus) pullRandomness() {
case <-con.resetRandomnessTicker:
case <-time.After(1500 * time.Millisecond):
// TODO(jimmy): pulling period should be related to lambdaBA.
- hashes := con.ccModule.pendingBlocksWithoutRandomness()
+ hashes := con.bcModule.pendingBlocksWithoutRandomness()
if len(hashes) > 0 {
con.logger.Debug(
"Calling Network.PullRandomness", "blocks", hashes)
@@ -1196,7 +1145,8 @@ func (con *Consensus) deliverBlock(b *types.Block) {
case con.resetDeliveryGuardTicker <- struct{}{}:
default:
}
- if err := con.db.UpdateBlock(*b); err != nil {
+ // TODO(mission): do we need to put block when confirmed now?
+ if err := con.db.PutBlock(*b); err != nil {
panic(err)
}
if err := con.db.PutCompactionChainTipInfo(
@@ -1209,13 +1159,13 @@ func (con *Consensus) deliverBlock(b *types.Block) {
if b.Position.Round == con.roundForNewConfig {
// Get configuration for the round next to next round. Configuration
// for that round should be ready at this moment and is required for
- // lattice module. This logic is related to:
+ // blockChain module. This logic is related to:
// - roundShift
// - notifyGenesisRound
futureRound := con.roundForNewConfig + 1
futureConfig := utils.GetConfigWithPanic(con.gov, futureRound, con.logger)
con.logger.Debug("Append Config", "round", futureRound)
- if err := con.lattice.AppendConfig(
+ if err := con.bcModule.appendConfig(
futureRound, futureConfig); err != nil {
con.logger.Debug("Unable to append config",
"round", futureRound,
@@ -1238,14 +1188,14 @@ func (con *Consensus) deliverFinalizedBlocks() error {
}
func (con *Consensus) deliverFinalizedBlocksWithoutLock() (err error) {
- deliveredBlocks := con.ccModule.extractBlocks()
+ deliveredBlocks := con.bcModule.extractBlocks()
con.logger.Debug("Last blocks in compaction chain",
- "delivered", con.ccModule.lastDeliveredBlock(),
- "pending", con.ccModule.lastPendingBlock())
+ "delivered", con.bcModule.lastDeliveredBlock(),
+ "pending", con.bcModule.lastPendingBlock())
for _, b := range deliveredBlocks {
con.deliverBlock(b)
+ go con.event.NotifyTime(b.Finalization.Timestamp)
}
- err = con.lattice.PurgeBlocks(deliveredBlocks)
return
}
@@ -1271,34 +1221,14 @@ func (con *Consensus) processBlockLoop() {
// processBlock is the entry point to submit one block to a Consensus instance.
func (con *Consensus) processBlock(block *types.Block) (err error) {
+ // Block processed by blockChain can be out-of-order. But the output from
+ // blockChain (deliveredBlocks) cannot, thus we need to protect the part
+ // below with writer lock.
con.lock.Lock()
defer con.lock.Unlock()
- // Block processed by lattice can be out-of-order. But the output of lattice
- // (deliveredBlocks) cannot.
- deliveredBlocks, err := con.lattice.ProcessBlock(block)
- if err != nil {
+ if err = con.bcModule.addBlock(block); err != nil {
return
}
- // Pass delivered blocks to compaction chain.
- for _, b := range deliveredBlocks {
- if b.IsFinalized() {
- if con.nonFinalizedBlockDelivered {
- panic(fmt.Errorf("attempting to skip finalized block: %s", b))
- }
- con.logger.Debug("skip delivery of finalized block",
- "block", b,
- "finalization-height", b.Finalization.Height)
- continue
- } else {
- // Mark that some non-finalized block delivered. After this flag
- // turned on, it's not allowed to deliver finalized blocks anymore.
- con.nonFinalizedBlockDelivered = true
- }
- if err = con.ccModule.processBlock(b); err != nil {
- return
- }
- go con.event.NotifyTime(b.Finalization.Timestamp)
- }
if err = con.deliverFinalizedBlocksWithoutLock(); err != nil {
return
}
@@ -1307,36 +1237,28 @@ func (con *Consensus) processBlock(block *types.Block) (err error) {
// processFinalizedBlock is the entry point for handling finalized blocks.
func (con *Consensus) processFinalizedBlock(block *types.Block) error {
- return con.ccModule.processFinalizedBlock(block)
+ return con.bcModule.processFinalizedBlock(block)
}
// PrepareBlock would setup header fields of block based on its ProposerID.
-func (con *Consensus) prepareBlock(b *types.Block,
- proposeTime time.Time) (err error) {
- if err = con.lattice.PrepareBlock(b, proposeTime); err != nil {
- return
+func (con *Consensus) proposeBlock(position types.Position) (
+ *types.Block, error) {
+ b, err := con.bcModule.proposeBlock(position, time.Now().UTC())
+ if err != nil {
+ return nil, err
}
con.logger.Debug("Calling Governance.CRS", "round", b.Position.Round)
crs := con.gov.CRS(b.Position.Round)
if crs.Equal(common.Hash{}) {
con.logger.Error("CRS for round is not ready, unable to prepare block",
"position", &b.Position)
- err = ErrCRSNotReady
- return
+ return nil, ErrCRSNotReady
}
- err = con.signer.SignCRS(b, crs)
- return
-}
-
-// PrepareGenesisBlock would setup header fields for genesis block.
-func (con *Consensus) PrepareGenesisBlock(b *types.Block,
- proposeTime time.Time) (err error) {
- if err = con.prepareBlock(b, proposeTime); err != nil {
- return
+ if err = con.signer.SignCRS(b, crs); err != nil {
+ return nil, err
}
- if len(b.Payload) != 0 {
- err = ErrGenesisBlockNotEmpty
- return
+ if b.IsGenesis() && len(b.Payload) != 0 {
+ return nil, ErrGenesisBlockNotEmpty
}
- return
+ return b, nil
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-23 21:16:01 +0800