Derek解讀Bytom原始碼-孤塊管理
作者:Derek
簡介
本章介紹bytom程式碼孤塊管理
作者使用MacOS作業系統,其他平臺也大同小異
Golang Version: 1.8
孤塊介紹
什麼是孤塊
當節點收到了一個有效的區塊,而在現有的主鏈中卻未找到它的父區塊,那麼這個區塊被認為是“孤塊”。父區塊是指當前區塊的PreviousBlockHash欄位指向上一區塊的hash值。
接收到的孤塊會被儲存在孤塊池中,直到它們的父區塊被節點收到。一旦收到了父區塊,節點就會將孤塊從孤塊池中取出,並且連線到它的父區塊,讓它作為區塊鏈的一部分。
孤塊出現的原因
當兩個或多個區塊在很短的時間間隔內被挖出來,節點有可能會以不同的順序接收到它們,這個時候孤塊現象就會出現。
我們假設有三個高度分別為100、101、102的塊,分別以102、101、100的顛倒順序被節點接收。此時節點將102、101放入到孤塊管理快取池中,等待彼此的父塊。當高度為100的區塊被同步進來時,會被驗證區塊和交易,然後儲存到區塊鏈上。這時會對孤塊快取池進行遞迴查詢,根據高度為100的區塊找到101的區塊並存儲到區塊鏈上,再根據高度為101的區塊找到102的區塊並存儲到區塊鏈上。
孤塊原始碼分析
孤塊管理快取池結構體
protocol/orphan_manage.go
type OrphanManage struct {
orphan map[bc.Hash]*types.Block
prevOrphans map [bc.Hash][]*bc.Hash
mtx sync.RWMutex
}
func NewOrphanManage() *OrphanManage {
return &OrphanManage{
orphan: make(map[bc.Hash]*types.Block),
prevOrphans: make(map[bc.Hash][]*bc.Hash),
}
}
- orphan 儲存孤塊,key為block hash,value為block結構體
- prevOrphans 儲存孤塊的父塊
- mtx 互斥鎖,保護map結構在多併發讀寫狀態下保持資料一致
新增孤塊到快取池
func (o *OrphanManage) Add(block *types.Block) {
blockHash := block.Hash()
o.mtx.Lock()
defer o.mtx.Unlock()
if _, ok := o.orphan[blockHash]; ok {
return
}
o.orphan[blockHash] = block
o.prevOrphans[block.PreviousBlockHash] = append(o.prevOrphans[block.PreviousBlockHash], &blockHash)
log.WithFields(log.Fields{"hash": blockHash.String(), "height": block.Height}).Info("add block to orphan")
}
當一個孤塊被新增到快取池中,還需要記錄該孤塊的父塊hash。用於父塊hash的查詢
查詢孤塊和父孤塊
func (o *OrphanManage) Get(hash *bc.Hash) (*types.Block, bool) {
o.mtx.RLock()
block, ok := o.orphan[*hash]
o.mtx.RUnlock()
return block, ok
}
func (o *OrphanManage) GetPrevOrphans(hash *bc.Hash) ([]*bc.Hash, bool) {
o.mtx.RLock()
prevOrphans, ok := o.prevOrphans[*hash]
o.mtx.RUnlock()
return prevOrphans, ok
}
刪除孤塊
func (o *OrphanManage) Delete(hash *bc.Hash) {
o.mtx.Lock()
defer o.mtx.Unlock()
block, ok := o.orphan[*hash]
if !ok {
return
}
delete(o.orphan, *hash)
prevOrphans, ok := o.prevOrphans[block.PreviousBlockHash]
if !ok || len(prevOrphans) == 1 {
delete(o.prevOrphans, block.PreviousBlockHash)
return
}
for i, preOrphan := range prevOrphans {
if preOrphan == hash {
o.prevOrphans[block.PreviousBlockHash] = append(prevOrphans[:i], prevOrphans[i+1:]...)
return
}
}
}
刪除孤塊的過程中,同時刪除父塊
孤塊處理邏輯
protocol/block.go
func (c *Chain) processBlock(block *types.Block) (bool, error) {
blockHash := block.Hash()
if c.BlockExist(&blockHash) {
log.WithFields(log.Fields{"hash": blockHash.String(), "height": block.Height}).Info("block has been processed")
return c.orphanManage.BlockExist(&blockHash), nil
}
if parent := c.index.GetNode(&block.PreviousBlockHash); parent == nil {
c.orphanManage.Add(block)
return true, nil
}
if err := c.saveBlock(block); err != nil {
return false, err
}
bestBlock := c.saveSubBlock(block)
// ...
}
processBlock函式處理block塊加入區塊鏈上之前的過程。
c.BlockExist判斷當前block塊是否存在於區塊鏈上或是否存在孤塊快取池中,如果存在則返回。
c.index.GetNode判斷block塊的父節點是否存在。如果在現有的主鏈中卻未找到它的父區塊則將block塊新增到孤塊快取池。
c.saveBlock走到了這一步說明,block父節點是存在於區塊鏈,則將block塊儲存到區塊鏈。該函式會驗證區塊和交易有效性。
saveSubBlock 程式碼如下:
func (c *Chain) saveSubBlock(block *types.Block) *types.Block {
blockHash := block.Hash()
prevOrphans, ok := c.orphanManage.GetPrevOrphans(&blockHash)
if !ok {
return block
}
bestBlock := block
for _, prevOrphan := range prevOrphans {
orphanBlock, ok := c.orphanManage.Get(prevOrphan)
if !ok {
log.WithFields(log.Fields{"hash": prevOrphan.String()}).Warning("saveSubBlock fail to get block from orphanManage")
continue
}
if err := c.saveBlock(orphanBlock); err != nil {
log.WithFields(log.Fields{"hash": prevOrphan.String(), "height": orphanBlock.Height}).Warning("saveSubBlock fail to save block")
continue
}
if subBestBlock := c.saveSubBlock(orphanBlock); subBestBlock.Height > bestBlock.Height {
bestBlock = subBestBlock
}
}
return bestBlock
}
saveSubBlock 在孤塊快取池中查詢是否存在當前區塊的下一個區塊。比如當前區塊高度為100,則在孤塊快取池中查詢是否有區塊高度為101的區塊。如果存在則將101區塊儲存到區塊鏈並從孤塊快取池中刪除該區塊。
saveSubBlock是一個遞迴函式的實現。目的是為了尋找最深葉子節點的遞迴方式。比如當前區塊高度為100的,遞迴查詢出高度為99、98、97等高度的區塊。