死磕以太坊源碼分析之MPT樹-下
文章以及資料請查看：https://github.com/blockchainGuide/
上篇主要介紹了以太坊中的MPT樹的原理丙唧，這篇主要會對MPT樹涉及的源碼進行拆解分析授帕。trie模塊主要有以下幾個文件：
|-encoding.go 主要講編碼之間的轉(zhuǎn)換
|-hasher.go 實現(xiàn)了從某個結(jié)點開始計算子樹的哈希的功能
|-node.go 定義了一個Trie樹中所有結(jié)點的類型和解析的代碼
|-sync.go 實現(xiàn)了SyncTrie對象的定義和所有方法
|-iterator.go 定義了所有枚舉相關(guān)接口和實現(xiàn)
|-secure_trie.go 實現(xiàn)了SecureTrie對象
|-proof.go 為key構(gòu)造一個merkle證明
|-trie.go Trie樹的增刪改查
|-database.go 對內(nèi)存中的trie樹節(jié)點進行引用計數(shù)
實現(xiàn)概覽
encoding.go
這個主要是講三種編碼（KEYBYTES encoding磅甩、HEX encoding、COMPACT encoding）的實現(xiàn)與轉(zhuǎn)換，trie中全程都需要用到這些介袜，該文件中主要實現(xiàn)了如下功能：
- hex編碼轉(zhuǎn)換為Compact編碼：
hexToCompact() - Compact編碼轉(zhuǎn)換為hex編碼：
compactToHex() - keybytes編碼轉(zhuǎn)換為Hex編碼：
keybytesToHex() - hex編碼轉(zhuǎn)換為keybytes編碼：
hexToKeybytes() - 獲取兩個字節(jié)數(shù)組的公共前綴的長度：
prefixLen()
func hexToCompact(hex []byte) []byte {
terminator := byte(0)
if hasTerm(hex) { //檢查是否有結(jié)尾為0x10 => 16
terminator = 1 //有結(jié)束標(biāo)記16說明是葉子節(jié)點
hex = hex[:len(hex)-1] //去除尾部標(biāo)記
}
buf := make([]byte, len(hex)/2+1) // 字節(jié)數(shù)組
buf[0] = terminator << 5 // 標(biāo)志byte為00000000或者00100000
//如果長度為奇數(shù)痹兜，添加奇數(shù)位標(biāo)志1蛮粮，并把第一個nibble字節(jié)放入buf[0]的低四位
if len(hex)&1 == 1 {
buf[0] |= 1 << 4 // 奇數(shù)標(biāo)志 00110000
buf[0] |= hex[0] // 第一個nibble包含在第一個字節(jié)中 0011xxxx
hex = hex[1:]
}
//將兩個nibble字節(jié)合并成一個字節(jié)
decodeNibbles(hex, buf[1:])
return buf
//compact編碼轉(zhuǎn)化為Hex編碼
func compactToHex(compact []byte) []byte {
base := keybytesToHex(compact)
base = base[:len(base)-1]
// apply terminator flag
// base[0]包括四種情況
// 00000000 擴展節(jié)點偶數(shù)位
// 00000001 擴展節(jié)點奇數(shù)位
// 00000010 葉子節(jié)點偶數(shù)位
// 00000011 葉子節(jié)點奇數(shù)位
// apply terminator flag
if base[0] >= 2 {
//如果是葉子節(jié)點益缎，末尾添加Hex標(biāo)志位16
base = append(base, 16)
}
// apply odd flag
//如果是偶數(shù)位，chop等于2然想，否則等于1
chop := 2 - base[0]&1
return base[chop:]
}
//compact編碼轉(zhuǎn)化為Hex編碼
func compactToHex(compact []byte) []byte {
base := keybytesToHex(compact)
base = base[:len(base)-1]
// apply terminator flag
// base[0]包括四種情況
// 00000000 擴展節(jié)點偶數(shù)位
// 00000001 擴展節(jié)點奇數(shù)位
// 00000010 葉子節(jié)點偶數(shù)位
// 00000011 葉子節(jié)點奇數(shù)位
// apply terminator flag
if base[0] >= 2 {
//如果是葉子節(jié)點莺奔，末尾添加Hex標(biāo)志位16
base = append(base, 16)
}
// apply odd flag
//如果是偶數(shù)位，chop等于2变泄，否則等于1
chop := 2 - base[0]&1
return base[chop:]
}
// 將十六進制的bibbles轉(zhuǎn)成key bytes令哟，這只能用于偶數(shù)長度的key
func hexToKeybytes(hex []byte) []byte {
if hasTerm(hex) {
hex = hex[:len(hex)-1]
}
if len(hex)&1 != 0 {
panic("can't convert hex key of odd length")
}
key := make([]byte, (len(hex)+1)/2)
decodeNibbles(hex, key)
return key
}
// 返回a和b的公共前綴的長度
func prefixLen(a, b []byte) int {
var i, length = 0, len(a)
if len(b) < length {
length = len(b)
}
for ; i < length; i++ {
if a[i] != b[i] {
break
}
}
return i
}
node.go
四種節(jié)點
node 接口分四種實現(xiàn): fullNode，shortNode妨蛹，valueNode屏富，hashNode，其中只有 fullNode 和 shortNode 可以帶有子節(jié)點蛙卤。
type (
fullNode struct {
Children [17]node // 分支節(jié)點
flags nodeFlag
}
shortNode struct { //擴展節(jié)點
Key []byte
Val node //可能指向葉子節(jié)點狠半，也可能指向分支節(jié)點。
flags nodeFlag
}
hashNode []byte
valueNode []byte // 葉子節(jié)點值颤难，但是該葉子節(jié)點最終還是會包裝在shortNode中
)
trie.go
Trie對象實現(xiàn)了MPT樹的所有功能典予，包括(key, value)對的增刪改查、計算默克爾哈希乐严，以及將整個樹寫入數(shù)據(jù)庫中。
iterator.go
nodeIterator提供了遍歷樹內(nèi)部所有結(jié)點的功能衣摩。其結(jié)構(gòu)如下：此結(jié)構(gòu)體是在trie.go定義的
type nodeIterator struct {
trie.NodeIterator
t *odrTrie
err error
}
里面包含了一個接口NodeIterator昂验，它的實現(xiàn)則是由iterator.go來提供的，其方法如下：
func (it *nodeIterator) Next(descend bool) bool
func (it *nodeIterator) Hash() common.Hash
func (it *nodeIterator) Parent() common.Hash
func (it *nodeIterator) Leaf() bool
func (it *nodeIterator) LeafKey() []byte
func (it *nodeIterator) LeafBlob() []byte
func (it *nodeIterator) LeafProof() [][]byte
func (it *nodeIterator) Path() []byte {}
func (it *nodeIterator) seek(prefix []byte) error
func (it *nodeIterator) peek(descend bool) (*nodeIteratorState, *int, []byte, error)
func (it *nodeIterator) nextChild(parent *nodeIteratorState, ancestor common.Hash) (*nodeIteratorState, []byte, bool)
func (it *nodeIterator) push(state *nodeIteratorState, parentIndex *int, path []byte)
func (it *nodeIterator) pop()
NodeIterator的核心是Next方法艾扮，每調(diào)用一次這個方法既琴，NodeIterator對象代表的當(dāng)前節(jié)點就會更新至下一個節(jié)點，當(dāng)所有結(jié)點遍歷結(jié)束泡嘴，Next方法返回false甫恩。
生成NodeIterator結(jié)口的方法有以下3種：
①：Trie.NodeIterator(start []byte)
通過start參數(shù)指定從哪個路徑開始遍歷，如果為nil則從頭到尾按順序遍歷酌予。
②：NewDifferenceIterator(a, b NodeIterator)
當(dāng)調(diào)用NewDifferenceIterator(a, b NodeIterator)后磺箕，生成的NodeIterator只遍歷存在于 b 但不存在于 a 中的結(jié)點。
③：NewUnionIterator(iters []NodeIterator)
當(dāng)調(diào)用NewUnionIterator(its []NodeIterator)后抛虫，生成的NodeIterator遍歷的結(jié)點是所有傳入的結(jié)點的合集松靡。
database.go
Database是trie模塊對真正數(shù)據(jù)庫的緩存層，其目的是對緩存的節(jié)點進行引用計數(shù)建椰，從而實現(xiàn)區(qū)塊的修剪功能雕欺。主要方法如下：
func NewDatabase(diskdb ethdb.KeyValueStore) *Database
func NewDatabaseWithCache(diskdb ethdb.KeyValueStore, cache int) *Database
func (db *Database) DiskDB() ethdb.KeyValueReader
func (db *Database) InsertBlob(hash common.Hash, blob []byte)
func (db *Database) insert(hash common.Hash, blob []byte, node node)
func (db *Database) insertPreimage(hash common.Hash, preimage []byte)
func (db *Database) node(hash common.Hash) node
func (db *Database) Node(hash common.Hash) ([]byte, error)
func (db *Database) preimage(hash common.Hash) ([]byte, error)
func (db *Database) secureKey(key []byte) []byte
func (db *Database) Nodes() []common.Hash
func (db *Database) Reference(child common.Hash, parent common.Hash)
func (db *Database) Dereference(root common.Hash)
func (db *Database) dereference(child common.Hash, parent common.Hash)
func (db *Database) Cap(limit common.StorageSize) error
func (db *Database) Commit(node common.Hash, report bool) error
security_trie.go
可以理解為加密了的trie的實現(xiàn)，ecurity_trie包裝了一下trie樹， 所有的key都轉(zhuǎn)換成keccak256算法計算的hash值屠列。同時在數(shù)據(jù)庫里面存儲hash值對應(yīng)的原始的key啦逆。
但是官方在代碼里也注釋了，這個代碼不穩(wěn)定笛洛，除了測試用例夏志，別的地方并沒有使用該代碼。
proof.go
- Prove()：根據(jù)給定的
key撞蜂，在trie中盲镶，將滿足key中最大長度前綴的路徑上的節(jié)點都加入到proofDb（隊列中每個元素滿足：未編碼的hash以及對應(yīng)rlp編碼后的節(jié)點） - VerifyProof()：驗證
proffDb中是否存在滿足輸入的hash，和對應(yīng)key的節(jié)點蝌诡，如果滿足溉贿，則返回rlp解碼后的該節(jié)點。
實現(xiàn)細節(jié)
Trie對象的增刪改查
①：Trie樹的初始化
如果root不為空浦旱，就通過resolveHash來加載整個Trie樹宇色，如果為空，就新建一個Trie樹颁湖。
func New(root common.Hash, db *Database) (*Trie, error) {
if db == nil {
panic("trie.New called without a database")
}
trie := &Trie{
db: db,
}
if root != (common.Hash{}) && root != emptyRoot {
rootnode, err := trie.resolveHash(root[:], nil)
if err != nil {
return nil, err
}
trie.root = rootnode
}
return trie, nil
}
②：Trie樹的插入
首先Trie樹的插入是個遞歸調(diào)用的過程宣蠕，它會從根開始找，一直找到合適的位置插入甥捺。
func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error)
參數(shù)說明：
- n: 當(dāng)前要插入的節(jié)點
- prefix: 當(dāng)前已經(jīng)處理完的key(節(jié)點共有的前綴)
- key: 未處理完的部分key抢蚀，完整的
key = prefix + key - value：需要插入的值
返回值說明：
- bool : 操作是否改變了Trie樹(dirty)
- Node :插入完成后的子樹的根節(jié)點
接下來就是分別對shortNode、fullNode镰禾、hashNode皿曲、nil 幾種情況進行說明。
2.1：節(jié)點為nil
空樹直接返回shortNode吴侦， 此時整顆樹的根就含有了一個shortNode節(jié)點屋休。
case nil:
return true, &shortNode{key, value, t.newFlag()}, nil
2.2 ：節(jié)點為shortNode
首先計算公共前綴，如果公共前綴就等于
key备韧，那么說明這兩個key是一樣的劫樟，如果value也一樣的(dirty == false)，那么返回錯誤织堂。如果沒有錯誤就更新
shortNode的值然后返回如果公共前綴不完全匹配叠艳，那么就需要把公共前綴提取出來形成一個獨立的節(jié)點(擴展節(jié)點),擴展節(jié)點后面連接一個
branch節(jié)點，branch節(jié)點后面看情況連接兩個short節(jié)點易阳。首先構(gòu)建一個branch節(jié)點(branch := &fullNode{flags: t.newFlag()}),然后再branch節(jié)點的Children位置調(diào)用t.insert插入剩下的兩個short節(jié)點
matchlen := prefixLen(key, n.Key)
if matchlen == len(n.Key) {
dirty, nn, err := t.insert(n.Val, append(prefix, key[:matchlen]...), key[matchlen:], value)
if !dirty || err != nil {
return false, n, err
}
return true, &shortNode{n.Key, nn, t.newFlag()}, nil
}
branch := &fullNode{flags: t.newFlag()}
var err error
_, branch.Children[n.Key[matchlen]], err = t.insert(nil, append(prefix, n.Key[:matchlen+1]...), n.Key[matchlen+1:], n.Val)
if err != nil {
return false, nil, err
}
_, branch.Children[key[matchlen]], err = t.insert(nil, append(prefix, key[:matchlen+1]...), key[matchlen+1:], value)
if err != nil {
return false, nil, err
}
if matchlen == 0 {
return true, branch, nil
}
return true, &shortNode{key[:matchlen], branch, t.newFlag()}, nil
2.3: 節(jié)點為fullNode
節(jié)點是fullNode(也就是分支節(jié)點)虑绵，那么直接往對應(yīng)的孩子節(jié)點調(diào)用insert方法,然后把對應(yīng)的孩子節(jié)點指向新生成的節(jié)點。
dirty, nn, err := t.insert(n.Children[key[0]], append(prefix, key[0]), key[1:], value)
if !dirty || err != nil {
return false, n, err
}
n = n.copy()
n.flags = t.newFlag()
n.Children[key[0]] = nn
return true, n, nil
2.4: 節(jié)點為hashnode
暫時還在數(shù)據(jù)庫中的節(jié)點闽烙，先調(diào)用 t.resolveHash(n, prefix)來加載到內(nèi)存翅睛，然后調(diào)用insert方法來插入声搁。
rn, err := t.resolveHash(n, prefix)
if err != nil {
return false, nil, err
}
dirty, nn, err := t.insert(rn, prefix, key, value)
if !dirty || err != nil {
return false, rn, err
}
return true, nn, nil
③：Trie樹查詢值
其實就是根據(jù)輸入的hash，找到對應(yīng)的葉子節(jié)點的數(shù)據(jù)捕发。主要看TryGet方法疏旨。
參數(shù)：
-
origNode：當(dāng)前查找的起始node位置 -
key：輸入要查找的數(shù)據(jù)的hash -
pos：當(dāng)前hash匹配到第幾位
func (t *Trie) tryGet(origNode node, key []byte, pos int) (value []byte, newnode node, didResolve bool, err error) {
switch n := (origNode).(type) {
case nil: //表示當(dāng)前trie是空樹
return nil, nil, false, nil
case valueNode: ////這就是我們要查找的葉子節(jié)點對應(yīng)的數(shù)據(jù)
return n, n, false, nil
case *shortNode: ////在葉子節(jié)點或者擴展節(jié)點匹配
if len(key)-pos < len(n.Key) || !bytes.Equal(n.Key, key[pos:pos+len(n.Key)]) {
return nil, n, false, nil
}
value, newnode, didResolve, err = t.tryGet(n.Val, key, pos+len(n.Key))
if err == nil && didResolve {
n = n.copy()
n.Val = newnode
}
return value, n, didResolve, err
case *fullNode://在分支節(jié)點匹配
value, newnode, didResolve, err = t.tryGet(n.Children[key[pos]], key, pos+1)
if err == nil && didResolve {
n = n.copy()
n.Children[key[pos]] = newnode
}
return value, n, didResolve, err
case hashNode: //說明當(dāng)前節(jié)點是輕節(jié)點，需要從db中獲取
child, err := t.resolveHash(n, key[:pos])
if err != nil {
return nil, n, true, err
}
value, newnode, _, err := t.tryGet(child, key, pos)
return value, newnode, true, err
...
}
didResolve用于判斷trie樹是否會發(fā)生變化扎酷，tryGet()只是用來獲取數(shù)據(jù)的檐涝，當(dāng)hashNode去db中獲取該node值后需要更新現(xiàn)有的trie，didResolve就會發(fā)生變化法挨。其他就是基本的遞歸查找樹操作谁榜。
④：Trie樹更新值
更新值，其實就是調(diào)用insert方法進行操作凡纳。
到此Trie樹的增刪改查就講解的差不多了窃植。
將節(jié)點寫入到Trie的內(nèi)存數(shù)據(jù)庫
如果要把節(jié)點寫入到內(nèi)存數(shù)據(jù)庫，需要序列化荐糜，可以先去了解下以太坊的Rlp編碼巷怜。這部分工作由trie.Commit()完成，當(dāng)trie.Commit(nil)暴氏，會執(zhí)行序列化和緩存等操作延塑，序列化之后是使用的Compact Encoding進行編碼，從而達到節(jié)省空間的目的答渔。
func (t *Trie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
if t.db == nil {
panic("commit called on trie with nil database")
}
hash, cached, err := t.hashRoot(t.db, onleaf)
if err != nil {
return common.Hash{}, err
}
t.root = cached
return common.BytesToHash(hash.(hashNode)), nil
}
上述代碼大概講了這些：
- 每次執(zhí)行
Commit()关带，該trie的cachegen就會加 1 -
Commit()方法返回的是trie.root所指向的node的hash（未編碼） - 其中的
hashRoot()方法目的是返回trie.root所指向的node的hash以及每個節(jié)點都帶有各自hash的trie樹的root。
//為每個node生成一個hash
func (t *Trie) hashRoot(db *Database, onleaf LeafCallback) (node, node, error) {
if t.root == nil {
return hashNode(emptyRoot.Bytes()), nil, nil
}
h := newHasher(onleaf)
defer returnHasherToPool(h)
return h.hash(t.root, db, true) //為每個節(jié)點生成一個未編碼的hash
}
而hashRoot的核心方法就是 h.hash沼撕，它返回了頭節(jié)點的hash以及每個子節(jié)點都帶有hash的頭節(jié)點（Trie.root指向它）宋雏，大致做了以下幾件事：
①：如果我們不存儲節(jié)點，而只是哈希端朵，則從緩存中獲取數(shù)據(jù)
if hash, dirty := n.cache(); hash != nil {
if db == nil {
return hash, n, nil
}
if !dirty {
switch n.(type) {
case *fullNode, *shortNode:
return hash, hash, nil
default:
return hash, n, nil
}
}
}
②：遞歸調(diào)用h.hashChildren，求出所有的子節(jié)點的hash值燃箭，再把原有的子節(jié)點替換成現(xiàn)在子節(jié)點的hash值
2.1:如果節(jié)點是shortNode
首先把collapsed.Key從Hex Encoding 替換成 Compact Encoding, 然后遞歸調(diào)用hash方法計算子節(jié)點的hash和cache冲呢，從而把子節(jié)點替換成了子節(jié)點的hash值
collapsed, cached := n.copy(), n.copy()
collapsed.Key = hexToCompact(n.Key)
cached.Key = common.CopyBytes(n.Key)
if _, ok := n.Val.(valueNode); !ok {
collapsed.Val, cached.Val, err = h.hash(n.Val, db, false)
if err != nil {
return original, original, err
}
}
return collapsed, cached, nil
2.2:節(jié)點是fullNode
遍歷每個子節(jié)點，把子節(jié)點替換成子節(jié)點的Hash值招狸，否則的化這個節(jié)點沒有children敬拓。直接返回。
collapsed, cached := n.copy(), n.copy()
for i := 0; i < 16; i++ {
if n.Children[i] != nil {
collapsed.Children[i], cached.Children[i], err = h.hash(n.Children[i], db, false)
if err != nil {
return original, original, err
}
}
}
cached.Children[16] = n.Children[16]
return collapsed, cached, nil
③：存儲節(jié)點n的哈希值裙戏，如果我們指定了存儲層乘凸，它會寫對應(yīng)的鍵/值對
store()方法主要就做了兩件事：
-
rlp序列化collapsed節(jié)點并將其插入db磁盤中 - 生成當(dāng)前節(jié)點的
hash - 將節(jié)點哈希插入
db
3.1：空數(shù)據(jù)或者hashNode，則不處理
if _, isHash := n.(hashNode); n == nil || isHash {
return n, nil
}
3.2:生成節(jié)點的RLP編碼
h.tmp.Reset() // 緩存初始化
if err := rlp.Encode(&h.tmp, n); err != nil { //將當(dāng)前node序列化
panic("encode error: " + err.Error())
}
if len(h.tmp) < 32 && !force {
return n, nil // Nodes smaller than 32 bytes are stored inside their parent 編碼后的node長度小于32累榜，若force為true营勤，則可確保所有節(jié)點都被編碼
}
//長度過大的灵嫌，則都將被新計算出來的hash取代
hash, _ := n.cache() //取出當(dāng)前節(jié)點的hash
if hash == nil {
hash = h.makeHashNode(h.tmp) //生成哈希node
}
3.3:將Trie節(jié)點合并到中間內(nèi)存緩存中
hash := common.BytesToHash(hash)
db.lock.Lock()
db.insert(hash, h.tmp, n)
db.lock.Unlock()
// Track external references from account->storage trie
//跟蹤帳戶->存儲Trie中的外部引用
if h.onleaf != nil {
switch n := n.(type) {
case *shortNode:
if child, ok := n.Val.(valueNode); ok { //指向的是分支節(jié)點
h.onleaf(child, hash) //用于統(tǒng)計當(dāng)前節(jié)點的信息，比如當(dāng)前節(jié)點有幾個子節(jié)點葛作，當(dāng)前有效的節(jié)點數(shù)
}
case *fullNode:
for i := 0; i < 16; i++ {
if child, ok := n.Children[i].(valueNode); ok {
h.onleaf(child, hash)
}
}
}
}
到此為止將節(jié)點寫入到Trie的內(nèi)存數(shù)據(jù)庫就已經(jīng)完成了寿羞。
如果覺得文章不錯可以關(guān)注公眾號：區(qū)塊鏈技術(shù)棧，詳細的所有以太坊源碼分析文章內(nèi)容以及代碼資料都在其中赂蠢。
Trie樹緩存機制
Trie樹的結(jié)構(gòu)里面有兩個參數(shù)绪穆， 一個是cachegen,一個是cachelimit。這兩個參數(shù)就是cache控制的參數(shù)虱岂。 Trie樹每一次調(diào)用Commit方法玖院，會導(dǎo)致當(dāng)前的cachegen增加1。
func (t *Trie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
...
t.cachegen++
...
}
然后在Trie樹插入的時候第岖，會把當(dāng)前的cachegen存放到節(jié)點中难菌。
func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error) {
....
return true, &shortNode{n.Key, nn, t.newFlag()}, nil
}
func (t *Trie) newFlag() nodeFlag {
return nodeFlag{dirty: true, gen: t.cachegen}
}
如果 trie.cachegen - node.cachegen > cachelimit，就可以把節(jié)點從內(nèi)存里面拿掉绍傲。 也就是說節(jié)點經(jīng)過幾次Commit扔傅，都沒有修改，那么就把節(jié)點從內(nèi)存里面干掉烫饼。 只要trie路徑上新增或者刪除一個節(jié)點猎塞，整個路徑的節(jié)點都需要重新實例化杠纵，也就是節(jié)點中的nodeFlag被初始化了。都需要重新更新到db磁盤铝量。
拿掉節(jié)點過程在 hasher.hash方法中， 這個方法是在commit的時候調(diào)用银亲。如果方法的canUnload方法調(diào)用返回真慢叨，那么就拿掉節(jié)點务蝠，如果只返回了hash節(jié)點，而沒有返回node節(jié)點馏段，這樣節(jié)點就沒有引用轩拨，不久就會被gc清除掉。 節(jié)點被拿掉之后亡蓉，會用一個hashNode節(jié)點來表示這個節(jié)點以及其子節(jié)點喷舀。 如果后續(xù)需要使用淋肾，可以通過方法把這個節(jié)點加載到內(nèi)存里面來梯影。
func (h *hasher) hash(n node, db *Database, force bool) (node, node, error) {
....
// 從緩存中卸載節(jié)點。它的所有子節(jié)點將具有較低或相等的緩存世代號碼简识。
cacheUnloadCounter.Inc(1)
...
}
參考&總結(jié)
這部分重要的內(nèi)容也就上面講述的感猛，主要集中在Trie上面陪白，如果有不對的地方，可以及時指正哦咱士。
