Bytom側(cè)鏈Vapor源碼分析節(jié)點(diǎn)出塊過(guò)程是怎樣的

Bytom側(cè)鏈Vapor源碼分析節(jié)點(diǎn)出塊過(guò)程是怎樣的，相信很多沒(méi)有經(jīng)驗(yàn)的人對(duì)此束手無(wú)策，為此本文總結(jié)了問(wèn)題出現(xiàn)的原因和解決方法，通過(guò)這篇文章希望你能解決這個(gè)問(wèn)題。

小編將從Vapor節(jié)點(diǎn)的創(chuàng)建開始，進(jìn)而拓展講解Vapor節(jié)點(diǎn)出塊過(guò)程中所涉及的源碼。

下面對(duì)Vapor稍加介紹。Vapor是目前國(guó)內(nèi)主流公鏈Bytom的高性能側(cè)鏈，是從Bytom主鏈中發(fā)展出來(lái)的一條獨(dú)立的高性能側(cè)鏈。Vapor是平臺(tái)最重要的區(qū)塊鏈基礎(chǔ)設(shè)施之一，目前采用DPoS的共識(shí)算法，具有高性能、高安全、可擴(kuò)展等特點(diǎn)，用于搭建規(guī)模化的商業(yè)應(yīng)用。

Vapor節(jié)點(diǎn)創(chuàng)建及出塊模塊的啟動(dòng)

Vapor入口函數(shù)：

vapor/cmd/vapord/main.go

func main() {
	cmd := cli.PrepareBaseCmd(commands.RootCmd, "TM", os.ExpandEnv(config.DefaultDataDir()))
	cmd.Execute()
}

傳入?yún)?shù)node后會(huì)調(diào)用runNode函數(shù)并新建一個(gè)節(jié)點(diǎn)。

vapor/cmd/vapord/commands/run_node.go

func runNode(cmd *cobra.Command, args []string) error {
	startTime := time.Now()
	setLogLevel(config.LogLevel)

	// Create & start node
	n := node.NewNode(config)
	……
}

vapor節(jié)點(diǎn)的結(jié)構(gòu)：

vapor/node/node.go

type Node struct {
	cmn.BaseService

	config          *cfg.Config
	eventDispatcher *event.Dispatcher
	syncManager     *netsync.SyncManager

	wallet          *w.Wallet
	accessTokens    *accesstoken.CredentialStore
	notificationMgr *websocket.WSNotificationManager
	api             *api.API
	chain           *protocol.Chain
	blockProposer   *blockproposer.BlockProposer
	miningEnable    bool
}

其中與出塊和共識(shí)相關(guān)的是blockProposer字段

新建節(jié)點(diǎn)的部分源碼

vapor/node/node.go

func NewNode(config *cfg.Config) *Node {
	//……
	node := &Node{
		eventDispatcher: dispatcher,
		config:          config,
		syncManager:     syncManager,
		accessTokens:    accessTokens,
		wallet:          wallet,
		chain:           chain,
		miningEnable:    config.Mining,

		notificationMgr: notificationMgr,
	}

	node.blockProposer = blockproposer.NewBlockProposer(chain, accounts, txPool, dispatcher)
	node.BaseService = *cmn.NewBaseService(nil, "Node", node)
	return node
}

從這可以看到node.blockProposer本質(zhì)上是一個(gè)vapor的block生成器，實(shí)際控制node啟動(dòng)出塊的模塊是vapor/proposal/blockproposer/blockproposer.go中的：

func (b *BlockProposer) Start() {
	b.Lock()
	defer b.Unlock()

	// Nothing to do if the miner is already running
	if b.started {
		return
	}

	b.quit = make(chan struct{})
	go b.generateBlocks() //出塊功能的關(guān)鍵模塊

	b.started = true
	log.Infof("block proposer started")
}

出塊模塊可以通過(guò)api啟動(dòng)

vapor/api/miner.go

func (a *API) startMining() Response {
	a.blockProposer.Start()
	if !a.IsMining() {
		return NewErrorResponse(errors.New("Failed to start mining"))
	}
	return NewSuccessResponse("")
}

以上講解的是節(jié)點(diǎn)創(chuàng)建和出塊模塊啟動(dòng)所涉及的源碼。

從generateBlocks()函數(shù)開始，將要講解是Vapor出塊過(guò)程的具體源碼。

Vapor的出塊機(jī)制

Vapor采用的是DPoS的共識(shí)機(jī)制進(jìn)行出塊。DPoS是由被社區(qū)選舉的可信帳戶（受托人，得票數(shù)排行前10位）來(lái)創(chuàng)建區(qū)塊。為了成為正式受托人，用戶要去社區(qū)拉票，獲得足夠多用戶的信任。用戶根據(jù)自己持有的加密貨幣數(shù)量占總量的百分比來(lái)投票。DPoS機(jī)制類似于股份制公司，普通股民進(jìn)不了董事會(huì)，要投票選舉代表（受托人）代他們做決策。在講解Vapor的出塊流程之前，要先了解Vapor在DPoS的參數(shù)設(shè)定。

DPoS的參數(shù)信息位于 vapor/consensus/general.go

type DPOSConfig struct {
	NumOfConsensusNode      int64
	BlockNumEachNode        uint64
	RoundVoteBlockNums      uint64
	MinConsensusNodeVoteNum uint64
	MinVoteOutputAmount     uint64
	BlockTimeInterval       uint64
	MaxTimeOffsetMs         uint64
}

接下來(lái)對(duì)參數(shù)進(jìn)行具體解釋

• NumOfConsensusNode是DPOS中共識(shí)節(jié)點(diǎn)的數(shù)量，Vapor中設(shè)置為10，通過(guò)投票選出十個(gè)負(fù)責(zé)出塊的共識(shí)節(jié)點(diǎn)。

• BlockNumEachNode是每個(gè)共識(shí)節(jié)點(diǎn)連續(xù)出塊的數(shù)量，Vapor中設(shè)置為12。

• RoundVoteBlockNums為每輪投票的出塊數(shù)，Vapor中設(shè)置為1200，也就是說(shuō)每輪投票產(chǎn)生的共識(shí)節(jié)點(diǎn)會(huì)負(fù)責(zé)出塊1200個(gè)。

• MinConsensusNodeVoteNum是成為共識(shí)節(jié)點(diǎn)要求的最小BTM數(shù)量（單位為neu，一億分之一BTM），Vapor中設(shè)置為100000000000000，也就是說(shuō)一個(gè)節(jié)點(diǎn)想成為共識(shí)節(jié)點(diǎn)，賬戶中至少需要存有100萬(wàn)BTM。

• MinVoteOutputAmoun為節(jié)點(diǎn)進(jìn)行投票所要求的最小BTM 數(shù)量（單位為neu），Vapor中設(shè)置為100000000，節(jié)點(diǎn)想要參與投票，賬戶中需要1BTM

• BlockTimeInterval為最短出塊時(shí)間間隔，Vapor每間隔0.5秒出一個(gè)塊。

• MaxTimeOffsetMs為塊時(shí)間允許比當(dāng)前時(shí)間提前的最大秒數(shù)，在Vapor中設(shè)置為2秒。

講完DPoS的參數(shù)設(shè)置后，就可以看看Vapor上出塊的核心代碼 generateBlocks

vapor/proposal/blockproposer/blockproposer.go

func (b *BlockProposer) generateBlocks() {
	xpub := config.CommonConfig.PrivateKey().XPub()
	xpubStr := hex.EncodeToString(xpub[:])
	ticker := time.NewTicker(time.Duration(consensus.ActiveNetParams.BlockTimeInterval) * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-b.quit:
			return
		case <-ticker.C:
		}
		//1
		bestBlockHeader := b.chain.BestBlockHeader()
		bestBlockHash := bestBlockHeader.Hash()
		now := uint64(time.Now().UnixNano() / 1e6)
		base := now
		if now < bestBlockHeader.Timestamp {
			base = bestBlockHeader.Timestamp
		}
		minTimeToNextBlock := consensus.ActiveNetParams.BlockTimeInterval - base%consensus.ActiveNetParams.BlockTimeInterval
		nextBlockTime := base + minTimeToNextBlock
		if (nextBlockTime - now) < consensus.ActiveNetParams.BlockTimeInterval/10 {
			nextBlockTime += consensus.ActiveNetParams.BlockTimeInterval
		}
		
        //2
		blocker, err := b.chain.GetBlocker(&bestBlockHash, nextBlockTime)
		……
		if xpubStr != blocker {
			continue
		}
		
        
        //3
		warnDuration := time.Duration(consensus.ActiveNetParams.BlockTimeInterval*warnTimeNum/warnTimeDenom) * time.Millisecond
		criticalDuration := time.Duration(consensus.ActiveNetParams.BlockTimeInterval*criticalTimeNum/criticalTimeDenom) * time.Millisecond
		block, err := proposal.NewBlockTemplate(b.chain, b.accountManager, nextBlockTime, warnDuration, criticalDuration)
		……
		//4
		isOrphan, err := b.chain.ProcessBlock(block)
		……
        //5
        log.WithFields(log.Fields{"module": logModule, "height": block.BlockHeader.Height, "isOrphan": isOrphan, "tx": len(block.Transactions)}).Info("proposer processed block")

		if err = b.eventDispatcher.Post(event.NewProposedBlockEvent{Block: *block}); err != nil {
			log.WithFields(log.Fields{"module": logModule, "height": block.BlockHeader.Height, "error": err}).Error("proposer fail on post block")
		}
	}
}

代碼經(jīng)過(guò)精簡(jiǎn)，省略了一些無(wú)關(guān)緊要的部分，并將重要的部分，分為5個(gè)模塊。

1. 計(jì)算并調(diào)整出塊的時(shí)間

2. 通過(guò)GetBlocker 獲取順序下一個(gè)block的公鑰，并與當(dāng)前塊比對(duì)，判斷當(dāng)前塊的出塊順序是否合法。

3. 通過(guò)b.chain.ProcessBlock根據(jù)模板生成了一個(gè)block。

4. 通過(guò)chain.ProcessBlock(block)嘗試把block加工處理后加到本機(jī)持有的區(qū)塊鏈上。

5. 使用logrus框架記錄新的塊，并像網(wǎng)絡(luò)中廣播。

b.chain.GetBlocker

針對(duì)generateBlocks()中幾個(gè)重要的模塊進(jìn)行拆分講解。

vapor/protocol/consensus_node_manager.go

GetBlocker()傳入當(dāng)前高度塊的哈希和下一個(gè)塊的出塊時(shí)間。

// 返回一個(gè)特定時(shí)間戳的Blocker
func (c *Chain) GetBlocker(prevBlockHash *bc.Hash, timeStamp uint64) (string, error) {
    consensusNodeMap, err := c.getConsensusNodes(prevBlockHash)
	//……

	prevVoteRoundLastBlock, err := c.getPrevRoundLastBlock(prevBlockHash)
	//……
    
	startTimestamp := prevVoteRoundLastBlock.Timestamp + consensus.ActiveNetParams.BlockTimeInterval
	//獲取order，xpub為公鑰
	order := getBlockerOrder(startTimestamp, timeStamp, uint64(len(consensusNodeMap)))
	for xPub, consensusNode := range consensusNodeMap {
		if consensusNode.Order == order {
			return xPub, nil
		}
	}
	//……
}

• 通過(guò)調(diào)用c.getConsensusNodes()獲得一個(gè)存儲(chǔ)共識(shí)節(jié)點(diǎn)的Map。

• 獲取上一輪投票的最后一個(gè)塊，在加上最短出塊時(shí)間間隔，計(jì)算得到這一輪的開始時(shí)間戳。

• 調(diào)用getBlockerOrder，通過(guò)開始時(shí)間戳和當(dāng)前要出塊的時(shí)間戳計(jì)算出這個(gè)時(shí)間點(diǎn)出塊的order。

• 最后比對(duì)consensusNodeMap中consensusNode.Order，并返回公鑰。

這個(gè)模塊是為了找出當(dāng)前時(shí)間戳對(duì)應(yīng)出塊的共識(shí)節(jié)點(diǎn)，并返回節(jié)點(diǎn)的公鑰。因?yàn)镈PoS中出塊的節(jié)點(diǎn)和順序必須是固定的，而使用generateBlocks()模塊嘗試出塊的共識(shí)節(jié)點(diǎn)不一定是當(dāng)前時(shí)間的合法出塊節(jié)點(diǎn)，因此需要本模塊通過(guò)對(duì)比公鑰進(jìn)行節(jié)點(diǎn)資格的驗(yàn)證。

proposal.NewBlockTemplate

vapor/proposal/proposal.go

func NewBlockTemplate(chain *protocol.Chain, accountManager *account.Manager, timestamp uint64, warnDuration, criticalDuration time.Duration) (*types.Block, error) {
	builder := newBlockBuilder(chain, accountManager, timestamp, warnDuration, criticalDuration)
	return builder.build()
}

func newBlockBuilder(chain *protocol.Chain, accountManager *account.Manager, timestamp uint64, warnDuration, criticalDuration time.Duration) *blockBuilder {
	preBlockHeader := chain.BestBlockHeader()
	block := &types.Block{
		BlockHeader: types.BlockHeader{
			Version:           1,
			Height:            preBlockHeader.Height + 1,
			PreviousBlockHash: preBlockHeader.Hash(),
			Timestamp:         timestamp,
			BlockCommitment:   types.BlockCommitment{},
			BlockWitness:      types.BlockWitness{Witness: make([][]byte, consensus.ActiveNetParams.NumOfConsensusNode)},
		},
	}

	builder := &blockBuilder{
		chain:             chain,
		accountManager:    accountManager,
		block:             block,
		txStatus:          bc.NewTransactionStatus(),
		utxoView:          state.NewUtxoViewpoint(),
		warnTimeoutCh:     time.After(warnDuration),
		criticalTimeoutCh: time.After(criticalDuration),
		gasLeft:           int64(consensus.ActiveNetParams.MaxBlockGas),
		timeoutStatus:     timeoutOk,
	}
	return builder
}

在Vapor上每個(gè)區(qū)塊有區(qū)塊頭和區(qū)塊的主體，區(qū)塊頭中包含版本號(hào)、高度、上一區(qū)塊的hash、時(shí)間戳等等，主體包括區(qū)塊鏈的引用模塊、賬戶管理器、區(qū)塊頭、Transaction狀態(tài)（版本號(hào)和驗(yàn)證狀態(tài)）、utxo視圖等。這一部分的目的是將，區(qū)塊的各種信息通過(guò)模板包裝成一個(gè)block交給后面的ProcessBlock(block)加工處理。

b.chain.ProcessBlock

vapor/protocol/block.go

func (c *Chain) ProcessBlock(block *types.Block) (bool, error) {
	reply := make(chan processBlockResponse, 1)
	c.processBlockCh <- &processBlockMsg{block: block, reply: reply}
	response := <-reply
	return response.isOrphan, response.err
}

func (c *Chain) blockProcesser() {
	for msg := range c.processBlockCh {
		isOrphan, err := c.processBlock(msg.block)
		msg.reply <- processBlockResponse{isOrphan: isOrphan, err: err}
	}
}

很顯然，這只是鏈更新的入口，block數(shù)據(jù)通過(guò)processBlockMsg結(jié)構(gòu)傳入了c.processBlockCh這個(gè)管道。隨后數(shù)據(jù)通過(guò)blockProcesser()處理后存入了msg.reply管道，而最后處理這個(gè)block的是processBlock()函數(shù)：

func (c *Chain) processBlock(block *types.Block) (bool, error) {
	//1
	blockHash := block.Hash()
	if c.BlockExist(&blockHash) {
		log.WithFields(log.Fields{"module": logModule, "hash": blockHash.String(), "height": block.Height}).Debug("block has been processed")
		return c.orphanManage.BlockExist(&blockHash), nil
	}
	//2
	c.markTransactions(block.Transactions...)
	//3
	if _, err := c.store.GetBlockHeader(&block.PreviousBlockHash); err != nil {
		c.orphanManage.Add(block)
		return true, nil
	}
	//4
	if err := c.saveBlock(block); err != nil {
		return false, err
	}
	
	bestBlock := c.saveSubBlock(block)
	bestBlockHeader := &bestBlock.BlockHeader

	c.cond.L.Lock()
	defer c.cond.L.Unlock()
	//5
	if bestBlockHeader.PreviousBlockHash == c.bestBlockHeader.Hash() {
		log.WithFields(log.Fields{"module": logModule}).Debug("append block to the end of mainchain")
		return false, c.connectBlock(bestBlock)
	}
	//6
	if bestBlockHeader.Height > c.bestBlockHeader.Height {
		log.WithFields(log.Fields{"module": logModule}).Debug("start to reorganize chain")
		return false, c.reorganizeChain(bestBlockHeader)
	}
	return false, nil
}

processBlock()函數(shù)返回的bool表示的是block是否為孤塊。

1. 通過(guò)block的hash判斷這個(gè)block是否已經(jīng)在鏈上。若已存在，則報(bào)錯(cuò)并返回false（表示該block不是孤塊）

2. 將block中的Transactions標(biāo)記，后續(xù)會(huì)調(diào)用c.knownTxs.Add()將Transactions加入到Transaction集合中。

3. 判斷是否為孤塊，如果是，則調(diào)用孤塊管理部分的模塊處理并返回true。

4. 保存block，在saveBlock()中會(huì)對(duì)簽名和區(qū)塊進(jìn)行驗(yàn)證。

5. bestBlockHeader.PreviousBlockHash == c.bestBlockHeader.Hash()的情況說(shuō)明一切正常，新block被添加到鏈的末端。

6. bestBlockHeader.Height > c.bestBlockHeader.Height 表示出現(xiàn)了分叉，需要回滾。

看完上述內(nèi)容，你們掌握Bytom側(cè)鏈Vapor源碼分析節(jié)點(diǎn)出塊過(guò)程是怎樣的的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！