以太坊DPOS共識機制源碼分析

本文小編為大家詳細介紹“以太坊DPOS共識機制源碼分析”，內容詳細，步驟清晰，細節處理妥當，希望這篇“以太坊DPOS共識機制源碼分析”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來學習新知識吧。

一、前言：

任何共識機制都必須回答包括但不限于如下的問題：

1. 下一個添加到數據庫的新區塊應該由誰來生成？

2. 下一個塊應該何時產生？

3. 該區塊應包含哪些交易？

4. 怎樣對本協議進行修改？

5. 該如何解決交易歷史的競爭問題？

二、功能描述：

每一個持有比特股的人按照持股比例進行對候選受托人的投票；從中選取投票數最多的前21位代表(也可以是其他數字，具體由區塊鏈項目方決定) 成為權力完全相等的21個超級節點（真正：受托人/見證人）；通過每隔3秒輪詢方式產出區塊；而其他候選受托人無權產出區塊；

1、持股人：比特股持有所有人；每個賬戶按照持幣數給證人投票；可以隨時更換投票；也可以不投；但投只能投贊成票；

2、見證人（受托人/代表，類似比特幣的礦工）：

注冊成為候選受托人需要支付一筆保證金，這筆保證金就是為了防止節點出現作惡的情況，一般來說，如果成為受托人，也就成為超級節點進行挖礦，超級節點需要大概兩周的時間才能達到損益平衡，這就促使超級節點至少挖滿兩周不作惡。

3、選定代表（實現步驟中未考慮實現它）

 代表也是通過類似選舉證人的方式選舉出來。 創始賬戶（the genesis account）有權對網絡參數提出修改，而代表就是該特殊賬戶的共同簽署者。這些參數包括交易費用，區塊大小，證人工資和區塊間隔等。 在大多數代表批準了提議的變更后，股東有2周的復審期（review period），在此期間他們可以投票踢出代表并作廢被提議的變更。

4.出塊規則：每隔3秒輪詢受托人；并且每個證人會輪流地在固定的預先計劃好的2秒內生產一個區塊。 當所有證人輪完之后，將被洗牌。 如果某個證人沒有在他自己的時間段內生產一個區塊，那么該時間段將被跳過，下一個證人繼續產生下一個區塊。每當證人生產一個區塊時，他們都會獲取相應的服務費。 證人的薪酬水平由股東選出的代表（delegate）來制定。 如果某個證人沒有生產出區塊，那么就不會給他支付薪酬，并可能在未來被投票踢出。

5.算法主要包含兩個核心部分:塊驗證人選舉和塊驗證人調度

（1）第一批塊驗證人由創世塊指定，后續每個周期(周期由具體實現定義)都會在周期開始的第一個塊重新選舉。驗證人選舉過程如下:

1. 踢掉上個周期出塊不足的驗證人

2. 統計截止選舉塊(每個周期的第一塊)產生時候選人的票數，選出票數最高的前 N 個作為驗證人

3. 隨機打亂驗證人出塊順序，驗證人根據隨機后的結果順序出塊

（2）驗證人調度根據選舉結果進行出塊，其他節點根據選舉結果驗證出塊順序和選舉結果是否一致，不一致則認為此塊不合法，直接丟棄。

三、以太坊DPOS共識機制源碼分析

1、啟動入口：

以太坊入口調試腳本：

以太坊項目的啟動：main.go中的init()函數-->調用geth方法-->調用startNode-->backend.go中的函數StartMining-->miner.go中的start函數

func init() {
   // Initialize the CLI app and start Geth
   app.Action = geth
}

func geth(ctx *cli.Context) error {
   //根據上下文配置信息獲取全量節點并將該節點注冊到以太坊服務
   //makeFullNode函數-->flags.go中函數RegisterEthService中-->eth.New-->handler.go中NewProtocolManager-->InsertChain內進行dpos的區塊信息校驗
   node := makeFullNode(ctx)
   //啟動節點
   startNode(ctx, node)
   node.Wait()
   return nil
}

啟動節點說明

func startNode(ctx *cli.Context, stack *node.Node) {
   //啟動當前節點：utils.StartNode(stack)
  //解鎖注冊錢包事件并自動派生錢包
  //監聽錢包事件
   if ctx.GlobalBool(utils.MiningEnabledFlag.Name) {
      //判斷是否全量節點，只有全量節點才有挖礦權利：
      var ethereum *eth.Ethereum
      if err := stack.Service(&ethereum); err != nil {
         utils.Fatalf("ethereum service not running: %v", err)
      }
      //設置gas價格
      ethereum.TxPool().SetGasPrice(utils.GlobalBig(ctx, utils.GasPriceFlag.Name))
      //驗證是否當前出塊受托人：validator, err := s.Validator() ，啟動服務打包區塊
      if err := ethereum.StartMining(true); err != nil {
         utils.Fatalf("Failed to start mining: %v", err)
      }
   }
}

上面StartMining方法會調用miner.go中的start函數，調用啟動函數之前已經啟動全量節點，并進行相關初始化工作（具體初始化內容如下）；

func (self *Miner) Start(coinbase common.Address) {
   atomic.StoreInt32(&self.shouldStart, 1)
   self.worker.setCoinbase(coinbase)
   self.coinbase = coinbase

   if atomic.LoadInt32(&self.canStart) == 0 {
      log.Info("Network syncing, will start miner afterwards")
      return
   }
   atomic.StoreInt32(&self.mining, 1)

   log.Info("Starting mining operation")
   //獲取當前節點地址，啟動服務
   self.worker.start()
}

worker.go的start函數調用mintLoop函數

func (self *worker) mintLoop() {
   ticker := time.NewTicker(time.Second).C
   //for循環不斷監聽self信號，當監測到self停止時，則調用關閉操作代碼，并直接挑出循環監聽，函數退出。
   for {
      select {
      case now := <-ticker:
         self.mintBlock(now.Unix())//打包塊
      case <-self.stopper:
         close(self.quitCh)
         self.quitCh = make(chan struct{}, 1)
         self.stopper = make(chan struct{}, 1)
         return
      }
   }
}

2.相關角色說明

dpos_context.go

type DposContext struct {
   epochTrie     *trie.Trie  //記錄每個周期的驗證人列表
   delegateTrie  *trie.Trie  //記錄驗證人以及對應投票人的列表
   voteTrie      *trie.Trie //記錄投票人對應驗證人
   candidateTrie *trie.Trie //記錄候選人列表
   mintCntTrie   *trie.Trie //記錄驗證人在周期內的出塊數目
   db ethdb.Database 
}

以太坊MPT（Trie樹， Patricia Trie， 和Merkle樹）樹形結構存儲，并定期同步[k,v]型底層數據庫是LevelDB數據庫

3.相關交易類型說明

以太坊DPOS共識算法中，將"成為候選人"、"退出候選人"、"投票（授權）"、"取消投票（取消授權）"等操作均定義為以太坊的一種交易類型

transaction.go

const (//交易類型
   Binary TxType = iota  //之前的交易主要是轉賬或者合約調用
   LoginCandidate  //成為候選人
   LogoutCandidate  //退出候選人
   Delegate   //投票（授權）
   UnDelegate  //取消投票（取消授權）
)

在一個新塊打包時會執行所有塊內的交易，如果發現交易類型不是之前的轉賬或者合約調用類型，那么會調用 applyDposMessage 進行處理

在worker.go的createNewWork()-->commitTransactions函數-->commitTransaction函數-->調用state_processor.go中的ApplyTransaction函數-->applyDposMessage

func ApplyTransaction(config *params.ChainConfig, dposContext *types.DposContext, bc *BlockChain, author *common.Address, gp *GasPool, statedb *state.StateDB, header *types.Header, tx *types.Transaction, usedGas *big.Int, cfg vm.Config) (*types.Receipt, *big.Int, error) {
   msg, err := tx.AsMessage(types.MakeSigner(config, header.Number))
   if err != nil {
      return nil, nil, err
   }

   if msg.To() == nil && msg.Type() != types.Binary {
      return nil, nil, types.ErrInvalidType
   }

   // 創建EVM環境的上下文
   context := NewEVMContext(msg, header, bc, author)
   // Create a new environment which holds all relevant information
   // 創建EVM虛擬機處理交易及智能合約
   vmenv := vm.NewEVM(context, statedb, config, cfg)
   // Apply the transaction to the current state (included in the env)
   _, gas, failed, err := ApplyMessage(vmenv, msg, gp)
   if err != nil {
      return nil, nil, err
   }
   if msg.Type() != types.Binary {
      //如果是非轉賬或者合約調用類型交易
      if err = applyDposMessage(dposContext, msg); err != nil {
         return nil, nil, err
      }
   }
}

func applyDposMessage(dposContext *types.DposContext, msg types.Message) error {
   switch msg.Type() {
   case types.LoginCandidate://成為候選人
      dposContext.BecomeCandidate(msg.From())
   case types.LogoutCandidate://取消候選人
      dposContext.KickoutCandidate(msg.From())
   case types.Delegate://投票
      //投票之前需要先檢查該賬號是否候選人；如果投票人之前已經給其他人投過票則先取消之前投票，再進行投票
      dposContext.Delegate(msg.From(), *(msg.To()))
   case types.UnDelegate://取消投票
      dposContext.UnDelegate(msg.From(), *(msg.To()))
   default:
      return types.ErrInvalidType
   }
   return nil
}

4.打包出塊過程

worker.go  

func (self *worker) mintBlock(now int64) {
   engine, ok := self.engine.(*dpos.Dpos)
   if !ok {
      log.Error("Only the dpos engine was allowed")
      return
   }
   //礦工會定時（每隔3秒）檢查當前的 validator 是否為當前節點，如果是則說明輪詢到自己出塊了；
   err := engine.CheckValidator(self.chain.CurrentBlock(), now)
   if err != nil {
      switch err {
      case dpos.ErrWaitForPrevBlock,
         dpos.ErrMintFutureBlock,
         dpos.ErrInvalidBlockValidator,
         dpos.ErrInvalidMintBlockTime:
         log.Debug("Failed to mint the block, while ", "err", err)
      default:
         log.Error("Failed to mint the block", "err", err)
      }
      return
   }
   //創建一個新的打塊任務
   work, err := self.createNewWork()
   if err != nil {
      log.Error("Failed to create the new work", "err", err)
      return
   }
   //Seal 會對新塊進行簽名
   result, err := self.engine.Seal(self.chain, work.Block, self.quitCh)
   if err != nil {
      log.Error("Failed to seal the block", "err", err)
      return
   }
   //將新塊廣播到鄰近的節點，其他節點接收到新塊會根據塊的簽名以及選舉結果來看新塊是否應該由該驗證人來出塊
   self.recv <- &Result{work, result}
}

func (self *worker) createNewWork() (*Work, error) {
   //......

   num := parent.Number()
   header := &types.Header{
      ParentHash: parent.Hash(),
      Number:     num.Add(num, common.Big1),
      GasLimit:   core.CalcGasLimit(parent),
      GasUsed:    new(big.Int),
      Extra:      self.extra,
      Time:       big.NewInt(tstamp),
   }
   // 僅在挖掘時設置coinbase（避免偽塊獎勵）
   if atomic.LoadInt32(&self.mining) == 1 {
      header.Coinbase = self.coinbase
   }
   //初始化塊頭基礎信息
   if err := self.engine.Prepare(self.chain, header); err != nil {
      return nil, fmt.Errorf("got error when preparing header, err: %s", err)
   }
   
   //主要是從 transaction pool 按照 gas price 將交易打包到塊中
   txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
   work.commitTransactions(self.mux, txs, self.chain, self.coinbase)

   // 打包區塊
   var (
      uncles    []*types.Header
      badUncles []common.Hash
   )
   for hash, uncle := range self.possibleUncles {
      if len(uncles) == 2 {
         break
      }
      if err := self.commitUncle(work, uncle.Header()); err != nil {
         log.Trace("Bad uncle found and will be removed", "hash", hash)
         log.Trace(fmt.Sprint(uncle))

         badUncles = append(badUncles, hash)
      } else {
         log.Debug("Committing new uncle to block", "hash", hash)
         uncles = append(uncles, uncle.Header())
      }
   }
   for _, hash := range badUncles {
      delete(self.possibleUncles, hash)
   }
   // 將 prepare 和 CommitNewWork 內容打包成新塊，同時里面還有包含出塊獎勵、選舉、更新打塊計數等功能
   if work.Block, err = self.engine.Finalize(self.chain, header, work.state, work.txs, uncles, work.receipts, work.dposContext); err != nil {
      return nil, fmt.Errorf("got error when finalize block for sealing, err: %s", err)
   }
   work.Block.DposContext = work.dposContext
   return work, nil
}

疑問：

（1）.這里面沒看到跟pow一樣的工作量難度證明的哈希函數計算，即當有出塊權益時，打包驗證好交易后是否直接打包，那如何會出現規定時間打包失敗的情況呢？是否是只有類似斷網或網絡不好時會出現？

（2）.Seal會對新塊進行封裝簽名；在pow算法中seal是核心是計算工作量得出隨機符合條件hash，而在dpos共識中seal是否只做了封裝簽名操作？從源碼中看是這樣

5.選舉分析

（1）選舉實現步驟：

1. 根據上個周期出塊的情況把一些被選上但出塊數達不到要求的候選人踢掉

2. 截止到上一塊為止，選出票數最高的前 N 個候選人作為驗證人

3. 打亂驗證人順序

當調用dpos.go中Finalize函數打包新塊時

func (d *Dpos) Finalize(chain consensus.ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction,
   uncles []*types.Header, receipts []*types.Receipt, dposContext *types.DposContext) (*types.Block, error) {
   // 累積塊獎勵并提交最終狀態根
   AccumulateRewards(chain.Config(), state, header, uncles)
   header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number))

   parent := chain.GetHeaderByHash(header.ParentHash)
   epochContext := &EpochContext{
      statedb:     state,
      DposContext: dposContext,
      TimeStamp:   header.Time.Int64(),
   }
   if timeOfFirstBlock == 0 {
      if firstBlockHeader := chain.GetHeaderByNumber(1); firstBlockHeader != nil {
         timeOfFirstBlock = firstBlockHeader.Time.Int64()
      }
   }
   genesis := chain.GetHeaderByNumber(0)
   //打包每個塊之前調用 tryElect 來看看當前塊是否是新周期的第一塊，如果是第一塊則需要觸發選舉。
   err := epochContext.tryElect(genesis, parent)
   if err != nil {
      return nil, fmt.Errorf("got error when elect next epoch, err: %s", err)
   }

   //更新驗證人在周期內的出塊數目
   updateMintCnt(parent.Time.Int64(), header.Time.Int64(), header.Validator, dposContext)
   header.DposContext = dposContext.ToProto()
   return types.NewBlock(header, txs, uncles, receipts), nil
}

epoch_context.go中的tryElect選舉函數

func (ec *EpochContext) tryElect(genesis, parent *types.Header) error {
   genesisEpoch := genesis.Time.Int64() / epochInterval
   prevEpoch := parent.Time.Int64() / epochInterval
   currentEpoch := ec.TimeStamp / epochInterval

   prevEpochIsGenesis := prevEpoch == genesisEpoch
   if prevEpochIsGenesis && prevEpoch < currentEpoch {
      prevEpoch = currentEpoch - 1
   }

   prevEpochBytes := make([]byte, 8)
   binary.BigEndian.PutUint64(prevEpochBytes, uint64(prevEpoch))
   iter := trie.NewIterator(ec.DposContext.MintCntTrie().PrefixIterator(prevEpochBytes))
   //根據當前塊和上一塊的時間計算當前塊和上一塊是否屬于同一周期，如果是同一周期，意味著當前塊不是周期第一塊，不需要觸發選舉；如果不是同一周期，說明當前塊是該周期的第一塊，則觸發選舉
   for i := prevEpoch; i < currentEpoch; i++ {
      // 如果前一個周期不是創世周期，觸發踢出候選人規則；
      if !prevEpochIsGenesis && iter.Next() {
         //踢出規則主要看上一周期是否存在候選人出塊少于特定閾值（50%），如果存在則踢出：if cnt < epochDuration/blockInterval/ maxValidatorSize /2 {
         if err := ec.kickoutValidator(prevEpoch); err != nil {
            return err
         }
      }
      //對候選人進行計票
      votes, err := ec.countVotes()
      if err != nil {
         return err
      }
      candidates := sortableAddresses{}
      for candidate, cnt := range votes {
         candidates = append(candidates, &sortableAddress{candidate, cnt})
      }
      if len(candidates) < safeSize {
         return errors.New("too few candidates")
      }
      //將候選人按照票數由高到低排序
      sort.Sort(candidates)
      if len(candidates) > maxValidatorSize {//如果候選人大于預定受托人數量常量maxValidatorSize，則選出前maxValidatorSize個為受托人
         candidates = candidates[:maxValidatorSize]
      }

      // 重排受托人，由于使用seed是由父塊的hash以及當前周期編號組成，所以每個節點計算出來的受托人列表也會一致；
      seed := int64(binary.LittleEndian.Uint32(crypto.Keccak512(parent.Hash().Bytes()))) + i
      r := rand.New(rand.NewSource(seed))
      for i := len(candidates) - 1; i > 0; i-- {
         j := int(r.Int31n(int32(i + 1)))
         candidates[i], candidates[j] = candidates[j], candidates[i]
      }
      sortedValidators := make([]common.Address, 0)
      for _, candidate := range candidates {
         sortedValidators = append(sortedValidators, candidate.address)
      }
      //保存受托人列表
      epochTrie, _ := types.NewEpochTrie(common.Hash{}, ec.DposContext.DB())
      ec.DposContext.SetEpoch(epochTrie)
      ec.DposContext.SetValidators(sortedValidators)
      log.Info("Come to new epoch", "prevEpoch", i, "nextEpoch", i+1)
   }
   return nil
}

（2）計票實現

1. 先找出候選人對應投票人的列表

2. 所有投票人的余額作為票數累積到候選人的總票數中

計票實現函數是epoch_context.go中的tryElect選舉函數中的countVotes函數

func (ec *EpochContext) countVotes() (votes map[common.Address]*big.Int, err error) {
   votes = map[common.Address]*big.Int{}
   delegateTrie := ec.DposContext.DelegateTrie()//記錄驗證人以及對應投票人的列表
   candidateTrie := ec.DposContext.CandidateTrie()//獲取候選人列表
   statedb := ec.statedb

   iterCandidate := trie.NewIterator(candidateTrie.NodeIterator(nil))
   existCandidate := iterCandidate.Next()
   if !existCandidate {
      return votes, errors.New("no candidates")
   }
   //遍歷候選人列表
   for existCandidate {
      candidate := iterCandidate.Value
      candidateAddr := common.BytesToAddress(candidate)
      delegateIterator := trie.NewIterator(delegateTrie.PrefixIterator(candidate))
      existDelegator := delegateIterator.Next()
      if !existDelegator {
         votes[candidateAddr] = new(big.Int)
         existCandidate = iterCandidate.Next()
         continue
      }
      //遍歷后續人對應的投票人列表
      for existDelegator {
         delegator := delegateIterator.Value
         score, ok := votes[candidateAddr]
         if !ok {
            score = new(big.Int)
         }
         delegatorAddr := common.BytesToAddress(delegator)
         //獲取投票人的余額作為票數累積到候選人的票數中
         weight := statedb.GetBalance(delegatorAddr)
         score.Add(score, weight)
         votes[candidateAddr] = score
         existDelegator = delegateIterator.Next()
      }
      existCandidate = iterCandidate.Next()
   }
   return votes, nil
}

讀到這里，這篇“以太坊DPOS共識機制源碼分析”文章已經介紹完畢，想要掌握這篇文章的知識點還需要大家自己動手實踐使用過才能領會，如果想了解更多相關內容的文章，歡迎關注億速云行業資訊頻道。