以太坊DPOS源碼分析

1、前言：

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

1. 下一個添加到數據庫的新區塊應該由誰來生成？
2. 下一個塊應該什麼時候產生？
3. 該區塊應包含哪些交易？
4. 怎樣對本協議進行修改？
5. 該如何解決交易歷史的競爭問題？

2、功能描述：

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

一、持股人：比特股持有全部人；每一個帳戶按照持幣數給證人投票；能夠隨時更換投票；也能夠不投；但投只能投同意票；

二、見證人（受託人/表明，相似比特幣的礦工）：

註冊成爲候選受託人須要支付一筆保證金，這筆保證金就是爲了防止節點出現做惡的狀況，通常來講，若是成爲受託人，也就成爲超級節點進行挖礦，超級節點須要大概兩週的時間才能達到損益平衡，這就促使超級節點至少挖滿兩週不做惡。

三、選定表明（實現步驟中未考慮實現它）

 表明也是經過相似選舉證人的方式選舉出來。 創始帳戶（the genesis account）有權對網絡參數提出修改，而表明就是該特殊帳戶的共同簽署者。這些參數包括交易費用，區塊大小，證人工資和區塊間隔等。 在大多數表明批准了提議的變動後，股東有2周的複審期（review period），在此期間他們能夠投票踢出表明並做廢被提議的變動。

4.出塊規則：每隔3秒輪詢受託人；而且每一個證人會輪流地在固定的預先計劃好的2秒內生產一個區塊。 當全部證人輪完以後，將被洗牌。 若是某個證人沒有在他本身的時間段內生產一個區塊，那麼該時間段將被跳過，下一個證人繼續產生下一個區塊。每當證人生產一個區塊時，他們都會獲取相應的服務費。 證人的薪酬水平由股東選出的表明（delegate）來制定。 若是某個證人沒有生產出區塊，那麼就不會給他支付薪酬，並可能在將來被投票踢出。

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

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

1. 踢掉上個週期出塊不足的驗證人
2. 統計截止選舉塊(每一個週期的第一塊)產生時候選人的票數，選出票數最高的前 N 個做爲驗證人
3. 隨機打亂驗證人出塊順序，驗證人根據隨機後的結果順序出塊

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

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

一、啓動入口：

以太坊入口調試腳本：

以太坊項目的啓動：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
}