move pool into chainmgr

Author: envestcc

blockchain/blockchain.go

@@ -116,7 +116,6 @@ type (
 	BlockBuilderFactory interface {
 		// NewBlockBuilder creates block builder
 		Mint(ctx context.Context, pk crypto.PrivateKey) (*block.Block, error)
-		Init(hash.Hash256)
 	}
 
 	// blockchain implements the Blockchain interface
@@ -253,17 +252,7 @@ func (bc *blockchain) Start(ctx context.Context) error {
 				GetBlockTime: bc.getBlockTime,
 			},
 		), bc.genesis))
-	err := bc.lifecycle.OnStart(ctx)
-	if err != nil {
-		return err
-	}
-	// init block builder factory
-	if tip, err := bc.tipInfo(bc.TipHeight()); err != nil {
-		return errors.Wrap(err, "failed to get tip info")
-	} else {
-		bc.bbf.Init(tip.Hash)
-		return nil
-	}
+	return bc.lifecycle.OnStart(ctx)
 }
 
 // Stop stops the blockchain.

consensus/scheme/rolldpos/chainmanager.go

@@ -62,16 +62,13 @@ type (
 	}
 	// StateReaderFactory is the factory interface of state reader
 	StateReaderFactory interface {
-		StateReaderAt(*block.Header) (protocol.StateReader, error)
+		StateReaderAt(hash.Hash256) (protocol.StateReader, error)
 	}
 
 	// BlockBuilderFactory is the factory interface of block builder
 	BlockBuilderFactory interface {
 		Mint(ctx context.Context, pk crypto.PrivateKey) (*block.Block, error)
 		ReceiveBlock(*block.Block) error
-		Init(hash.Hash256)
-		AddProposal(*block.Block) error
-		Block(hash.Hash256) *block.Block
 	}
 
 	chainManager struct {
@@ -80,6 +77,7 @@ type (
 		timerFactory *prometheustimer.TimerFactory
 		bc           blockchain.Blockchain
 		bbf          BlockBuilderFactory
+		pool         *proposalPool
 	}
 
 	forkChain struct {
@@ -118,6 +116,7 @@ func NewChainManager(bc blockchain.Blockchain, srf StateReaderFactory, bbf Block
 		bbf:          bbf,
 		timerFactory: timerFactory,
 		srf:          srf,
+		pool:         newProposalPool(),
 	}
 }
 
@@ -197,7 +196,7 @@ func (cm *chainManager) mintNewBlock(timestamp time.Time, pk crypto.PrivateKey,
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to create block")
 	}
-	if err = cm.bbf.AddProposal(blk); err != nil {
+	if err = cm.pool.AddBlock(blk); err != nil {
 		blkHash := blk.HashBlock()
 		log.L().Error("failed to add proposal", zap.Error(err), zap.Uint64("height", blk.Height()), log.Hex("hash", blkHash[:]))
 	}
@@ -221,11 +220,7 @@ func (fc *forkChain) TipHash() hash.Hash256 {
 }
 
 func (cm *chainManager) StateReader() (protocol.StateReader, error) {
-	head, err := cm.bc.BlockHeaderByHeight(cm.bc.TipHeight())
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to get the head block")
-	}
-	return cm.srf.StateReaderAt(head)
+	return cm.srf.StateReaderAt(cm.bc.TipHash())
 }
 
 // StateReader returns the state reader
@@ -239,7 +234,7 @@ func (cm *chainManager) ValidateBlock(blk *block.Block) error {
 	if err != nil {
 		return err
 	}
-	if err = cm.bbf.AddProposal(blk); err != nil {
+	if err = cm.pool.AddBlock(blk); err != nil {
 		blkHash := blk.HashBlock()
 		log.L().Error("failed to add proposal", zap.Error(err), zap.Uint64("height", blk.Height()), log.Hex("hash", blkHash[:]))
 	}
@@ -255,6 +250,10 @@ func (cm *chainManager) CommitBlock(blk *block.Block) error {
 		blkHash := blk.HashBlock()
 		log.L().Error("failed to receive block", zap.Error(err), zap.Uint64("height", blk.Height()), log.Hex("hash", blkHash[:]))
 	}
+	if err := cm.pool.ReceiveBlock(blk); err != nil {
+		blkHash := blk.HashBlock()
+		log.L().Error("failed to receive block", zap.Error(err), zap.Uint64("height", blk.Height()), log.Hex("hash", blkHash[:]))
+	}
 	return nil
 }
 
@@ -263,32 +262,38 @@ func (cm *chainManager) Start(ctx context.Context) error {
 	if err != nil {
 		return errors.Wrap(err, "failed to get the head block")
 	}
-	cm.bbf.Init(head.HashBlock())
+	cm.pool.Init(head.HashBlock())
 	return nil
 }
 
 // Fork creates a new chain manager with the given hash
 func (cm *chainManager) Fork(hash hash.Hash256) (ForkChain, error) {
-	head, err := cm.bc.BlockHeaderByHeight(cm.bc.TipHeight())
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to get the head block")
-	}
-	if hash != head.HashBlock() {
-		blk := cm.bbf.Block(hash)
+	var (
+		head *block.Header
+		err  error
+		tip  = cm.tipInfo()
+	)
+	if hash != tip.Hash {
+		blk := cm.pool.BlockByHash(hash)
 		if blk == nil {
 			return nil, errors.Errorf("block %x not found when fork", hash)
 		}
 		head = &blk.Header
+	} else {
+		head, err = cm.bc.BlockHeaderByHeight(tip.Height)
+		if head == nil {
+			return nil, errors.Errorf("block %x not found when fork", hash)
+		}
 	}
-	sr, err := cm.srf.StateReaderAt(head)
+	sr, err := cm.srf.StateReaderAt(hash)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to create state reader at %d, hash %x", head.Height(), head.HashBlock())
 	}
 	return newForkChain(cm, head, sr), nil
 }
 
 func (cm *chainManager) draftBlockByHeight(height uint64, tipHash hash.Hash256) *block.Block {
-	for blk := cm.bbf.Block(tipHash); blk != nil && blk.Height() >= height; blk = cm.bbf.Block(blk.PrevHash()) {
+	for blk := cm.pool.BlockByHash(tipHash); blk != nil && blk.Height() >= height; blk = cm.pool.BlockByHash(blk.PrevHash()) {
 		if blk.Height() == height {
 			return blk
 		}
@@ -305,6 +310,10 @@ func (cm *chainManager) TipHash() hash.Hash256 {
 }
 
 func (fc *forkChain) tipHash() hash.Hash256 {
+	if fc.head.Height() == 0 {
+		g := fc.cm.bc.Genesis()
+		return g.Hash()
+	}
 	return fc.head.HashBlock()
 }
 

state/factory/proposalpool.go renamed to consensus/scheme/rolldpos/proposalpool.go

@@ -1,4 +1,4 @@
-package factory
+package rolldpos
 
 import (
 	"sync"
@@ -14,30 +14,30 @@ import (
 
 // proposalPool is a pool of draft blocks
 type proposalPool struct {
-	// blocks is a map of draft blocks
+	// nodes is a map of draft proposal blocks
 	// key is the hash of the block
-	blocks map[hash.Hash256]*block.Block
-	// forks is a map of draft blocks that are forks
+	nodes map[hash.Hash256]*block.Block
+	// leaves is a map of tip blocks of forks
 	// key is the hash of the tip block of the fork
 	// value is the timestamp of the block
-	forks map[hash.Hash256]time.Time
+	leaves map[hash.Hash256]time.Time
 	// root is the hash of the tip block of the blockchain
 	root hash.Hash256
 	mu   sync.Mutex
 }
 
 func newProposalPool() *proposalPool {
 	return &proposalPool{
-		blocks: make(map[hash.Hash256]*block.Block),
-		forks:  make(map[hash.Hash256]time.Time),
+		nodes:  make(map[hash.Hash256]*block.Block),
+		leaves: make(map[hash.Hash256]time.Time),
 	}
 }
 
 func (d *proposalPool) Init(root hash.Hash256) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	d.root = root
-	log.L().Info("proposal pool initialized", log.Hex("root", root[:]))
+	log.L().Debug("proposal pool initialized", log.Hex("root", root[:]))
 }
 
 // AddBlock adds a block to the draft pool
@@ -46,19 +46,19 @@ func (d *proposalPool) AddBlock(blk *block.Block) error {
 	defer d.mu.Unlock()
 	// nothing to do if the block already exists
 	hash := blk.HashBlock()
-	if _, ok := d.blocks[hash]; ok {
+	if _, ok := d.nodes[hash]; ok {
 		return nil
 	}
 	// it must be a new tip of any fork, or make a new fork
 	prevHash := blk.PrevHash()
-	if _, ok := d.forks[prevHash]; ok {
-		delete(d.forks, prevHash)
+	if _, ok := d.leaves[prevHash]; ok {
+		delete(d.leaves, prevHash)
 	} else if prevHash != d.root {
 		return errors.Errorf("block %x is not a tip of any fork", prevHash[:])
 	}
-	d.forks[hash] = blk.Timestamp()
-	d.blocks[hash] = blk
-	log.L().Info("added block to draft pool", log.Hex("hash", hash[:]), zap.Uint64("height", blk.Height()), zap.Time("timestamp", blk.Timestamp()))
+	d.leaves[hash] = blk.Timestamp()
+	d.nodes[hash] = blk
+	log.L().Debug("added block to draft pool", log.Hex("hash", hash[:]), zap.Uint64("height", blk.Height()), zap.Time("timestamp", blk.Timestamp()))
 	return nil
 }
 
@@ -74,22 +74,22 @@ func (d *proposalPool) ReceiveBlock(blk *block.Block) error {
 	}
 
 	// remove blocks in other forks or older blocks in the same fork
-	for f := range d.forks {
-		start := d.blocks[f]
+	for f := range d.leaves {
+		start := d.nodes[f]
 		if f == fork {
 			start = blk
 		}
-		for b := start; b != nil; b = d.blocks[b.PrevHash()] {
+		for b := start; b != nil; b = d.nodes[b.PrevHash()] {
 			ha := b.HashBlock()
-			log.L().Info("remove block from draft pool", log.Hex("hash", ha[:]), zap.Uint64("height", b.Height()), zap.Time("timestamp", b.Timestamp()))
-			delete(d.blocks, b.HashBlock())
+			log.L().Debug("remove block from draft pool", log.Hex("hash", ha[:]), zap.Uint64("height", b.Height()), zap.Time("timestamp", b.Timestamp()))
+			delete(d.nodes, b.HashBlock())
 		}
 	}
 	// reset forks to only this one
-	if forkTime, ok := d.forks[fork]; ok && d.blocks[fork] != nil {
-		d.forks = map[hash.Hash256]time.Time{fork: forkTime}
+	if forkTime, ok := d.leaves[fork]; ok && d.nodes[fork] != nil {
+		d.leaves = map[hash.Hash256]time.Time{fork: forkTime}
 	} else {
-		d.forks = map[hash.Hash256]time.Time{}
+		d.leaves = map[hash.Hash256]time.Time{}
 	}
 	d.root = blk.HashBlock()
 	return nil
@@ -100,7 +100,7 @@ func (d *proposalPool) Block(height uint64) *block.Block {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	var blk *block.Block
-	for _, b := range d.blocks {
+	for _, b := range d.nodes {
 		if b.Height() != height {
 			continue
 		} else if blk == nil {
@@ -116,13 +116,13 @@ func (d *proposalPool) Block(height uint64) *block.Block {
 func (d *proposalPool) BlockByHash(hash hash.Hash256) *block.Block {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	return d.blocks[hash]
+	return d.nodes[hash]
 }
 
 func (d *proposalPool) BlockByTime(prevHash hash.Hash256, timestamp time.Time) *block.Block {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	for _, b := range d.blocks {
+	for _, b := range d.nodes {
 		if b.PrevHash() == prevHash && b.Timestamp().Equal(timestamp) {
 			return b
 		}
@@ -133,12 +133,12 @@ func (d *proposalPool) BlockByTime(prevHash hash.Hash256, timestamp time.Time) *
 func (d *proposalPool) forkAt(blk *block.Block) (hash.Hash256, error) {
 	blkHash := blk.HashBlock()
 	// If this block isn't in the pool, just return it
-	if _, ok := d.blocks[blkHash]; !ok {
+	if _, ok := d.nodes[blkHash]; !ok {
 		return blkHash, nil
 	}
 	// Otherwise, find which fork chain contains it
-	for forkTip := range d.forks {
-		for b := d.blocks[forkTip]; b != nil; b = d.blocks[b.PrevHash()] {
+	for forkTip := range d.leaves {
+		for b := d.nodes[forkTip]; b != nil; b = d.nodes[b.PrevHash()] {
 			if blkHash == b.HashBlock() {
 				return forkTip, nil
 			}

consensus/scheme/rolldpos/rolldpos.go

@@ -162,12 +162,7 @@ func (r *RollDPoS) Calibrate(height uint64) {
 // ValidateBlockFooter validates the signatures in the block footer
 func (r *RollDPoS) ValidateBlockFooter(blk *block.Block) error {
 	height := blk.Height()
-	prevHash := blk.PrevHash()
-	fork, err := r.ctx.Chain().Fork(prevHash)
-	if err != nil {
-		return errors.Wrapf(err, "failed to get fork at height %d with prevHash %x", height, prevHash)
-	}
-	roundCalc := r.ctx.RoundCalculator().Fork(fork)
+	roundCalc := r.ctx.RoundCalculator().Fork(r.ctx.Chain())
 	round, err := roundCalc.NewRound(height, r.ctx.BlockInterval(height), blk.Timestamp(), nil)
 	if err != nil {
 		return err

consensus/scheme/rolldpos/rolldpos_test.go

@@ -201,8 +201,8 @@ func TestValidateBlockFooter(t *testing.T) {
 	bc.EXPECT().BlockFooterByHeight(blockHeight).Return(footer, nil).AnyTimes()
 	bc.EXPECT().ChainID().Return(uint32(1)).AnyTimes()
 	bc.EXPECT().TipHeight().Return(blockHeight).AnyTimes()
-	bc.EXPECT().BlockHeaderByHeight(blockHeight).Return(&block.Header{}, nil).Times(6)
-	bc.EXPECT().TipHash().Return(hash.ZeroHash256).Times(1)
+	bc.EXPECT().BlockHeaderByHeight(blockHeight).Return(&block.Header{}, nil).AnyTimes()
+	bc.EXPECT().TipHash().Return(hash.ZeroHash256).AnyTimes()
 	sk1 := identityset.PrivateKey(1)
 	g := genesis.TestDefault()
 	g.NumDelegates = 4

e2etest/local_test.go

@@ -559,8 +559,8 @@ func TestStartExistingBlockchain(t *testing.T) {
 
 func newTestConfig() (config.Config, error) {
 	cfg := config.Default
-	cfg.Genesis = genesis.TestDefault()
 	cfg = deepcopy.Copy(cfg).(config.Config)
+	cfg.Genesis = genesis.TestDefault()
 	cfg.Chain.TrieDBPath = _triePath
 	cfg.Chain.ChainDBPath = _dBPath
 	cfg.Chain.BlobStoreDBPath = _blobPath

state/factory/factory.go

@@ -88,7 +88,7 @@ type (
 		PutBlock(context.Context, *block.Block) error
 		WorkingSet(context.Context) (protocol.StateManager, error)
 		WorkingSetAtHeight(context.Context, uint64, ...*action.SealedEnvelope) (protocol.StateManager, error)
-		StateReaderAt(header *block.Header) (protocol.StateReader, error)
+		StateReaderAt(hash.Hash256) (protocol.StateReader, error)
 	}
 
 	// factory implements StateFactory interface, tracks changes to account/contract and batch-commits to DB
@@ -548,7 +548,7 @@ func (sf *factory) ReadView(name string) (interface{}, error) {
 }
 
 // StateReaderAt returns a state reader at a specific height
-func (sf *factory) StateReaderAt(header *block.Header) (protocol.StateReader, error) {
+func (sf *factory) StateReaderAt(hash.Hash256) (protocol.StateReader, error) {
 	panic("implement me")
 }