Derek解讀Bytom源碼-孤塊管理

Donne 發(fā)布于2019-06-27 18:39 / 1395人閱讀

摘要：此時(shí)節(jié)點(diǎn)將放入到孤塊管理緩存池中，等待彼此的父塊。判斷當(dāng)前塊是否存在于區(qū)塊鏈上或是否存在孤塊緩存池中，如果存在則返回。比如當(dāng)前區(qū)塊高度為，則在孤塊緩存池中查詢是否有區(qū)塊高度為的區(qū)塊。

作者：Derek

簡(jiǎn)介

Github地址：https://github.com/Bytom/bytom

Gitee地址：https://gitee.com/BytomBlockc...

本章介紹bytom代碼孤塊管理

作者使用MacOS操作系統(tǒng)，其他平臺(tái)也大同小異Golang Version: 1.8

孤塊介紹 什么是孤塊

當(dāng)節(jié)點(diǎn)收到了一個(gè)有效的區(qū)塊，而在現(xiàn)有的主鏈中卻未找到它的父區(qū)塊，那么這個(gè)區(qū)塊被認(rèn)為是“孤塊”。父區(qū)塊是指當(dāng)前區(qū)塊的PreviousBlockHash字段指向上一區(qū)塊的hash值。

接收到的孤塊會(huì)被存儲(chǔ)在孤塊池中，直到它們的父區(qū)塊被節(jié)點(diǎn)收到。一旦收到了父區(qū)塊，節(jié)點(diǎn)就會(huì)將孤塊從孤塊池中取出，并且連接到它的父區(qū)塊，讓它作為區(qū)塊鏈的一部分。

孤塊出現(xiàn)的原因

當(dāng)兩個(gè)或多個(gè)區(qū)塊在很短的時(shí)間間隔內(nèi)被挖出來(lái)，節(jié)點(diǎn)有可能會(huì)以不同的順序接收到它們，這個(gè)時(shí)候孤塊現(xiàn)象就會(huì)出現(xiàn)。

我們假設(shè)有三個(gè)高度分別為100、101、102的塊，分別以102、101、100的顛倒順序被節(jié)點(diǎn)接收。此時(shí)節(jié)點(diǎn)將102、101放入到孤塊管理緩存池中，等待彼此的父塊。當(dāng)高度為100的區(qū)塊被同步進(jìn)來(lái)時(shí)，會(huì)被驗(yàn)證區(qū)塊和交易，然后存儲(chǔ)到區(qū)塊鏈上。這時(shí)會(huì)對(duì)孤塊緩存池進(jìn)行遞歸查詢，根據(jù)高度為100的區(qū)塊找到101的區(qū)塊并存儲(chǔ)到區(qū)塊鏈上，再根據(jù)高度為101的區(qū)塊找到102的區(qū)塊并存儲(chǔ)到區(qū)塊鏈上。

孤塊源碼分析 孤塊管理緩存池結(jié)構(gòu)體

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 存儲(chǔ)孤塊，key為block hash，value為block結(jié)構(gòu)體

prevOrphans 存儲(chǔ)孤塊的父塊

mtx 互斥鎖，保護(hù)map結(jié)構(gòu)在多并發(fā)讀寫狀態(tài)下保持?jǐn)?shù)據(jù)一致

添加孤塊到緩存池

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")
}

當(dāng)一個(gè)孤塊被添加到緩存池中，還需要記錄該孤塊的父塊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
        }
    }
}

刪除孤塊的過(guò)程中，同時(shí)刪除父塊

孤塊處理邏輯

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函數(shù)處理block塊加入?yún)^(qū)塊鏈上之前的過(guò)程。

c.BlockExist判斷當(dāng)前block塊是否存在于區(qū)塊鏈上或是否存在孤塊緩存池中，如果存在則返回。

c.index.GetNode判斷block塊的父節(jié)點(diǎn)是否存在。如果在現(xiàn)有的主鏈中卻未找到它的父區(qū)塊則將block塊添加到孤塊緩存池。

c.saveBlock走到了這一步說(shuō)明，block父節(jié)點(diǎn)是存在于區(qū)塊鏈，則將block塊存儲(chǔ)到區(qū)塊鏈。該函數(shù)會(huì)驗(yàn)證區(qū)塊和交易有效性。

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 在孤塊緩存池中查詢是否存在當(dāng)前區(qū)塊的下一個(gè)區(qū)塊。比如當(dāng)前區(qū)塊高度為100，則在孤塊緩存池中查詢是否有區(qū)塊高度為101的區(qū)塊。如果存在則將101區(qū)塊存儲(chǔ)到區(qū)塊鏈并從孤塊緩存池中刪除該區(qū)塊。

saveSubBlock是一個(gè)遞歸函數(shù)的實(shí)現(xiàn)。目的是為了尋找最深葉子節(jié)點(diǎn)的遞歸方式。比如當(dāng)前區(qū)塊高度為100的，遞歸查詢出高度為99、98、97等高度的區(qū)塊。