以太坊源码分析--MPT树

摘要：它是以太坊中存储区块数据的核心数据结构。 它与树结构集成。 了解该结构对于学习以太坊区块模块源码和智能合约状态存储结构非常有帮助。

MPT（Merkle Patricia Tries）是以太坊中存储区块数据的核心数据结构。 它将 Merkle Tree 和 Patricia Tree 组合成一个树结构。 了解了MPT结构对之后，学习以太坊区块头和智能合约状态存储结构模块源码很有帮助。

先看默克尔树：

它的叶子是数据块的散列。 从图中可以看出，非叶子节点是其子节点串接后的哈希值。 底层数据的任何变化都会影响父节点。 这棵树的 Merkle Root 代表所有底层数据。 “概括”。

这样的树有很大的优势。 比如我们把交易信息写入这样一个树结构中，当需要证明一个交易是否存在于这棵树中时，就不需要重新计算所有交易的哈希值。 比如要证明图中的Hash 1-1以太坊区块结构，我们可以用Hash 1-0重新计算Hash 1，然后用Hash 0重新计算Top Hash，这样根据计算出的Top Hash是否和原来的一样Top Hash，如果相同，则Hash 1-1属于这棵树。

那么想象一下，我们把这个Top Hash存储在区块头中，那么有了区块头，就可以验证区块信息了。 同时Hash的计算过程可以非常快，可以在很短的时间内完成预处理。 利用 Merkle 树结构可以带来巨大的比较性能提升。

让我们再看看帕特里夏树：

Patricia树的特点从它的名字压缩前缀树结合上图就可以猜到。 这种树结构优于普通的以每个字符为节点的trie树结构。 它的键值可以使用多个字符。 ，这样可以降低树的高度，节省空间，我们再看一个例子：

数据中存储的键值对很容易在图中看到：

6c0a5c71ec20bq3w => 5

6c0a5c71ec20CX7j => 27

6c0a5c71781a1FXq => 18

6c0a5c71781a9狗 => 64

6c0a8f743b95zUfe => 30

6c0a8f743b95jx5R => 2

6c0a8f740d16y03G => 43

6c0a8f740d16vcc1 => 48

以太坊中的 MPT：

以太坊中MPT节点的规格主要包括以下内容：

NULL空节点，简单的表示为空，在代码中就是一个空串

Nibble 是密钥的基本单位，是一个四元组（四位的组合，比如0010用二进制表示就是一个四元组）

Extension节点有两个元素，一个是key值以太坊区块结构，一个是hash值，指向下一个节点

Branch 节点有 17 个元素。 回到Nibble，四元组是key的基本单位，四元组最多有16个值。 所以前16个在其遍历中一定落入key的16个可能的半字节值中的每一个。 第17个是存放那些以当前节点结束的节点（比如有3个key，分别是(abc, abd, ab)第17个字段存放的是ab节点的值）

Leaf叶子节点只有两个元素，即key和value

还有一些知识点需要了解。 为了将MPT树存入数据库，从数据库中恢复MPT树，对Extension和Leaf的节点类型做了特殊定义：如果是extension节点，那么前缀为0，而这个在密钥前面添加 0。 如果是叶子节点，则前缀为1。 同时，密钥的长度也针对校验类型进行了设置。 如果是奇数长度，则标记为1，如果是偶数长度，则标记为0。

以太坊中有几个地方使用了 MPT 树结构：

区块头中的 State Trie 状态树

key => sha3(以太坊账户地址)

value => rlp(账户内容信息账户)

Transactions Trie 区块头中的交易树

key => rlp (transaction offset 事务索引)

每个区块都有自己的交易树，不能更改

Receipts Trie 区块头中的收据树

key = rlp(transaction offset 交易索引)

每个区块都有自己的交易树，不能更改

Storage Trie 存储树

仅存储合约状态

每个账户都有自己的 Storage Trie

在这两个区块头中，state root，tx root receipt root分别存储了三棵树的根，第二个区块显示当账户175的数据发生变化时（27 -> 45），只需要存储部分数据与该账户相关，旧区块中的数据仍然可以正常访问。

MPT树种还有一个重要的概念，一种特殊的十六进制前缀（hex-prefix，HP）编码来对key进行编码。 我们先了解一下编码定义规则，后面再分析源码实现：

RAW原始编码，不对输入做任何改动

HEX 十六进制编码

每个RAW编码输入的字符被分解为高4位和低4位

如果是叶子节点，则在末尾添加Hex值0x10表示结束

如果是分支节点，不附加任何Hex值

比如key=>"bob",b的ASCII十六进制编码为0x62，o的ASCII十六进制编码为0x6f，分解成高四位和第四位，16表示终结 0x10，最终编码结果为[6 2 6 15 6 2 16]，

HEX-Prefix 十六进制前缀编码

输入key的结尾是0x10，然后去掉这个结束符

在键前添加一个四元数。 该Byte的第0位区分校验信息，第1位区分节点类型。

如果输入key的长度是偶数，则在flag 4-tuple后面再增加一个4-tuple 0x0

压缩原始密钥内容，将分离出来的两个字节高四位和低四位合并

十六进制前缀编码相当于一个逆向的过程，比如输入的是[6 2 6 15 6 2 16]，根据第一个规则去掉终止符16。根据第二个规则key前补一个四元组，从右往左第一位为1表示叶子节点，从右往左第0位如果后面key的长度为偶数设置为0，奇数长度设置为1，那么四元组0010就是2。根据第三个规则，添加一个全0的补在后面，那么就是20.根据第三个规则内容压缩合并，那么结果就是[0x20 0x62 0x6f 0x62]

官方有一个详细结构的例子：

下面通过一个图例来加深对上述MPT规则的理解

键的十六进制键值

做

动词

狗

小狗

总督

硬币

马

骘

三种编码格式相互转换的代码实现

compact就是上面提到的HEX-Prefix，keybytes就是以完整的字节（8bit）存储的普通信息，hex是以半字节（4bit）存储信息的格式。

go-ethereum/trie/编码：

package trie
func hexToCompact(hex []byte) []byte {
    terminator := byte(0)
    if hasTerm(hex) { //检查是否有结尾为0x10 => 16
        terminator = 1 //有结束标记16说明是叶子节点
        hex = hex[:len(hex)-1] //去除尾部标记
    }
    buf := make([]byte, len(hex)/2+1) // 字节数组
    
    buf[0] = terminator << 5 // 标志byte为00000000或者00100000
    //如果长度为奇数，添加奇数位标志1，并把第一个nibble字节放入buf[0]的低四位
    if len(hex)&1 == 1 {

        buf[0] |= 1 << 4 // 奇数标志 00110000
        buf[0] |= hex[0] // 第一个nibble包含在第一个字节中 0011xxxx
        hex = hex[1:]
    }
    //将两个nibble字节合并成一个字节
    decodeNibbles(hex, buf[1:])
    return buf
}
//compact编码转化为Hex编码
func compactToHex(compact []byte) []byte {
    base := keybytesToHex(compact)
    base = base[:len(base)-1]
     // apply terminator flag
    // base[0]包括四种情况
    // 00000000 扩展节点偶数位
    // 00000001 扩展节点奇数位
    // 00000010 叶子节点偶数位
    // 00000011 叶子节点奇数位
    // apply terminator flag
    if base[0] >= 2 {
       //如果是叶子节点，末尾添加Hex标志位16
        base = append(base, 16)
    }
    // apply odd flag
    //如果是偶数位，chop等于2，否则等于1
    chop := 2 - base[0]&1
    return base[chop:]
}
// 将keybytes 转成十六进制
func keybytesToHex(str []byte) []byte {
    l := len(str)*2 + 1
     //将一个keybyte转化成两个字节
    var nibbles = make([]byte, l)
    for i, b := range str {
        nibbles[i*2] = b / 16
        nibbles[i*2+1] = b % 16
    }
    //末尾加入Hex标志位16
    nibbles[l-1] = 16
    return nibbles
}
// 将十六进制的bibbles转成key bytes，这只能用于偶数长度的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
}
func decodeNibbles(nibbles []byte, bytes []byte) {
    for bi, ni := 0, 0; ni < len(nibbles); bi, ni = bi+1, ni+2 {
        bytes[bi] = nibbles[ni]<<4 | nibbles[ni+1]
    }
}
// 返回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
}
// 十六进制key是否有结束标志符
func hasTerm(s []byte) bool {
    return len(s) > 0 && s[len(s)-1] == 16

}

以太坊中的MTP数据结构

以上分析了以太坊的密钥编码方式。 接下来我们看一下以太坊中MPT树的数据结构。 在分析trie的数据结构之前，我们先了解一下node的定义：

特里/node.go

type node interface {
    fstring(string) string
    cache() (hashNode, bool)
    canUnload(cachegen, cachelimit uint16) bool
}
type (
    fullNode struct {  //分支节点
        Children [17]node // Actual trie node data to encode/decode (needs custom encoder)
        flags    nodeFlag
    }
    shortNode struct { 
        Key   []byte
        Val   node
        flags nodeFlag
    }
    hashNode  []byte
    valueNode []byte
)

上面代码中定义了四个struct，分别是四种类型的节点：

fullNode -> 分支节点，它有一个节点数组成员变量Children，容量为17，数组中的前16个空位对应十六进制（hex）中的0-9a-f，所以对于每个子节点，根据其key 16进制格式的第一个数字的值可以挂载到Children数组的某个位置，fullNode本身不需要额外的key变量； Children 数组的第 17 位是为 fullNode 的数据部分保留的。 这和我们上面提到的Branch节点的规格是一致的。

shortNode，key是任意长度的字符串（字节数组[]byte），体现了PatriciaTrie的特点。 通过将父节点及其子节点合并为只有一个子节点，缩短了 trie 的深度。 结果是某些节点将具有更长的密钥。

-> 扩展节点，Val指向分支节点或叶子节点

-> 叶子节点，val是rlp编码后的数据，key是数据的hash

valueNode -> MPT 的叶节点。 字节数组 []byte 的别名，没有孩子。 使用的valueNode是携带的数据部分的RLP哈希值，长度为32字节。 数据的RLP编码值作为valueNode的匹配项存储在数据库中。

hashNode -> 字符数组[]byte的别名，存放32byte的哈希值，即fullNode或shortNode对象的RLP哈希值

下面看看trie的结构和trie的运行。 使用上述节点的类型可能会更清楚。 我们看一下trie的结构定义

trie/trie.go：

type Trie struct {
    db           *Database // 用levelDB做KV存储
    root         node     //当前根节点
    originalRoot common.Hash //启动加载时候的hash，可以从db中恢复出整个trie
    cachegen, cachelimit uint16 // cachegen 缓存生成值，每次Commit会+1
}

这里的cachegen缓存生成值会附加到node节点上。 如果当前cachegen-cachelimit参数大于节点的缓存代数，则节点将从缓存中卸载以节省内存。 一个缓存多久不用就会从缓存中移除，这个看起来很像redis等一些LRU算法的cache db。

特里初始化：

func New(root common.Hash, db *Database) (*Trie, error) {
    if db == nil {
        panic("trie.New called without a database")
    }
    trie := &Trie{
        db:           db,
        originalRoot: root,
    }
    if (root != common.Hash{}) && root != emptyRoot {
       // 如果hash不是空值，从数据库中加载一个已经存在的树
        rootnode, err := trie.resolveHash(root[:], nil)
        if err != nil {
            return nil, err
        }
        trie.root = rootnode //根节点为找到的trie
    }
    //否则返回新建一个树
    return trie, nil
}

这里的trie.resolveHash是加载整个类树的方法，传入的root common.Hash hash是一个32位的byte[](common.HexToHash())将hex码转成原始hash。 让我们看看如何使用这个散列来找到整个 trie：

func (t *Trie) resolveHash(n hashNode, prefix []byte) (node, error) {
    cacheMissCounter.Inc(1) //没执行一次计数器+1
   //上面说过了，n是一个32位byte[]
    hash := common.BytesToHash(n)
  //通过hash从db中取出node的RLP编码内容
    enc, err := t.db.Node(hash)
    if err != nil || enc == nil {
        return nil, &MissingNodeError{NodeHash: hash, Path: prefix}
    }
    return mustDecodeNode(n, enc, t.cachegen), nil
}

mustDecodeNode 调用 decodeNode。 该方法通过RLP列表的长度判断编码后的内容属于上述节点。 如果是2个字段就是shortNode，如果是17个字段就是fullNode，然后调用各自的decode分析函数。

func decodeNode(hash, buf []byte, cachegen uint16) (node, error) {
    if len(buf) == 0 {
        return nil, io.ErrUnexpectedEOF
    }
    elems, _, err := rlp.SplitList(buf) //将buf拆分为列表的内容以及列表后的任何剩余字节。
    if err != nil {

        return nil, fmt.Errorf("decode error: %v", err)
    }
    switch c, _ := rlp.CountValues(elems); c {
    case 2:
        n, err := decodeShort(hash, elems, cachegen) //decode shortNode
        return n, wrapError(err, "short")
    case 17:
        n, err := decodeFull(hash, elems, cachegen) //decode fullNode
        return n, wrapError(err, "full")
    default:
        return nil, fmt.Errorf("invalid number of list elements: %v", c)
    }
}

decodeShort函数通过key是否包含结束标识来判断是叶子节点还是扩展节点。 我们已经在上面的编码部分提到了这一点。 如果有结束标识，则为叶子节点，然后通过rlp.SplitString解析val生成叶子。 节点 shortNode 被返回。 如果没有结束标识，则为扩展节点，由decodeRef解析，返回一个shortNode。

func decodeShort(hash, elems []byte, cachegen uint16) (node, error) {
    kbuf, rest, err := rlp.SplitString(elems) //将elems填入RLP字符串的内容以及字符串后的任何剩余字节。
    if err != nil {
        return nil, err
    }
    flag := nodeFlag{hash: hash, gen: cachegen}
    key := compactToHex(kbuf)
    if hasTerm(key) {
        // value node
        val, _, err := rlp.SplitString(rest)
        if err != nil {
            return nil, fmt.Errorf("invalid value node: %v", err)
        }
        return &shortNode{key, append(valueNode{}, val...), flag}, nil
    }
    r, _, err := decodeRef(rest, cachegen)
    if err != nil {
        return nil, wrapError(err, "val")
    }
    return &shortNode{key, r, flag}, nil
}

继续看decodeRef主要做了什么：

func decodeRef(buf []byte, cachegen uint16) (node, []byte, error) {
    kind, val, rest, err := rlp.Split(buf)
    if err != nil {
        return nil, buf, err
    }
    switch {
    case kind == rlp.List:
        // "embedded" node reference. The encoding must be smaller
        // than a hash in order to be valid.
        if size := len(buf) - len(rest); size > hashLen {
            err := fmt.Errorf("oversized embedded node (size is %d bytes, want size < %d)", size, hashLen)
            return nil, buf, err
        }
        n, err := decodeNode(nil, buf, cachegen)
        return n, rest, err
    case kind == rlp.String && len(val) == 0:
        // empty node
        return nil, rest, nil
    case kind == rlp.String && len(val) == 32:
        return append(hashNode{}, val...), rest, nil
    default:
        return nil, nil, fmt.Errorf("invalid RLP string size %d (want 0 or 32)", len(val))
    }
}

这段代码比较清晰。 rlp.Split 返回的类型的处理方式不同。 如果是列表，调用decodeNode解析。 如果是空节点，则返回空。 如果是32位哈希值，返回hashNode，decodeFull：

func decodeFull(hash, elems []byte, cachegen uint16) (*fullNode, error) {
    n := &fullNode{flags: nodeFlag{hash: hash, gen: cachegen}}
    for i := 0; i < 16; i++ {
        cld, rest, err := decodeRef(elems, cachegen)
        if err != nil {
            return n, wrapError(err, fmt.Sprintf("[%d]", i))
        }
        n.Children[i], elems = cld, rest
    }
    val, _, err := rlp.SplitString(elems)
    if err != nil {
        return n, err
    }
    if len(val) > 0 {
        n.Children[16] = append(valueNode{}, val...)
    }
    return n, nil

}

回到Trie结构中的cachegen，cachelimit，每次Trie树commit时cachegen都会+1。 这两个参数是缓存的控制参数。 为了弄清楚Trie的缓存机制，我们先看看Commit是干什么的。 的：

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
    t.cachegen++
    return common.BytesToHash(hash.(hashNode)), nil //返回所指向的node的未编码的hash
}
//返回trie.root所指向的node的hash以及每个节点都带有各自hash的trie树的root。
func (t *Trie) hashRoot(db *Database, onleaf LeafCallback) (node, node, error) {
    if t.root == nil {
        return hashNode(emptyRoot.Bytes()), nil, nil
    }
    h := newHasher(t.cachegen, t.cachelimit, onleaf)
    defer returnHasherToPool(h)
    return h.hash(t.root, db, true)//为每个节点生成一个未编码的hash
}

Commit的目的是将trie树中的key转换成Compact码，为每个节点生成一个hash，保证变化的数据能够正常提交到db。

那这个cachegen怎么放在节点中呢？ 当在节点中插入trie树时，会将当前trie的cachegen放入节点中。 查看trie的insert方法：

//n -> trie当前插入节点
//prefix -> 当前匹配到的key的公共前缀
//key -> 待插入数据当前key中剩余未匹配的部分，完整的key=prefix+key
//value -> 待插入数据本身
//返回 -> 是否改变树，插入完成后子树根节点，error
func (t *Trie) insert(n node, prefix, key []byte, value node) (bool, node, error) {
    if len(key) == 0 {
        if v, ok := n.(valueNode); ok {
            return !bytes.Equal(v, value.(valueNode)), value, nil
        }
        //如果key长度为0，那么说明当前节点中新增加的节点和当前节点数据一样，认为已经新增过了就直接返回
        return true, value, nil
    }
    switch n := n.(type) {
    case *shortNode:
        matchlen := prefixLen(key, n.Key) // 返回公共前缀长度
        
        if matchlen == len(n.Key) {
        //如果整个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
        }
        //如果它在索引0处出现则用该branch替换shortNode
        if matchlen == 0 {
            return true, branch, nil
        }
        // Otherwise, replace it with a short node leading up to the branch.
        return true, &shortNode{key[:matchlen], branch, t.newFlag()}, nil
    case *fullNode:
        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
    case nil:

    //在空trie中添加一个节点，就是叶子节点，返回shortNode。
        return true, &shortNode{key, value, t.newFlag()}, nil
    case hashNode:
        rn, err := t.resolveHash(n, prefix)//恢复一个存储在db中的node
        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
    default:
        panic(fmt.Sprintf("%T: invalid node: %v", n, n))
    }

Trie树的插入，是一种递归的调用方式，从根节点开始，向下查找，直到找到可以插入的点，进行插入操作。

如果当前根节点叶子节点shortNode，首先计算公共前缀

如果public prefix等于key，则两个key相同，如果value也相同（dirty == false），则返回错误。 如果没有错误，更新shortNode的值并返回。

如果公共前缀不完全匹配，则需要将公共前缀提取出来，形成一个独立的节点（扩展节点）。 扩展节点连接一个分支节点，分支节点视情况连接两个短节点。 先建一个分支节点(branch := &fullNode{flags: t.newFlag()})，然后在分支节点的Children位置调用t.insert插入剩下的两个短节点。 这里有个小细节，key的编码是HEX编码，最后有个终结符。 考虑我们的根节点的键是 abc0x16，我们插入的节点的键是 ab0x16。 下面的branch.Children[key[matchlen]]可以正常运行，0x16刚好指向分支节点的第17个子节点。 如果匹配长度为0，则直接返回分支节点，否则返回shortNode节点作为前缀节点。

如果节点类型为nil（一棵全新的Trie树的节点为nil），此时整棵树为空，直接返回shortNode{key, value, t.newFlag()}。 这时，整棵树的根节点就包含了一个shortNode节点。

如果当前节点是fullNode（即分支节点），那么直接调用对应子节点的insert方法，再将对应子节点指向新生成的节点。

如果当前节点是一个hashNode，hashNode表示当前节点还没有加载到内存中，或者存储在数据库中，那么先调用t.resolveHash(n, prefix)加载到内存中，然后调用insert在加载的节点方法上插入。

接下来我们看看如何遍历Trie树从Trie中获取数据，以及根据key获取value的过程：

func (t *Trie) TryGet(key []byte) ([]byte, error) {
    key = keybytesToHex(key)
    value, newroot, didResolve, err := t.tryGet(t.root, key, 0)
    if err == nil && didResolve {
        t.root = newroot
    }
    return value, err
}
func (t *Trie) tryGet(origNode node, key []byte, pos int) (value []byte, newnode node, didResolve bool, err error) {
    switch n := (origNode).(type) {
    case nil: 
       // 空树
        return nil, nil, false, nil
    case valueNode:
       // 就是要查找的叶子节点数据
        return n, n, false, nil
    case *shortNode: 
        if len(key)-pos < len(n.Key) || !bytes.Equal(n.Key, key[pos:pos+len(n.Key)]) {
            // key在trie中不存在
            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
            n.flags.gen = t.cachegen
        }
        return value, n, didResolve, err
    case *fullNode:
        value, newnode, didResolve, err = t.tryGet(n.Children[key[pos]], key, pos+1)
        if err == nil && didResolve {
            n = n.copy()
            n.flags.gen = t.cachegen
            n.Children[key[pos]] = newnode
        }
        return value, n, didResolve, err
    case hashNode:
       // hashNodes时候需要去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
    default:
        panic(fmt.Sprintf("%T: invalid node: %v", origNode, origNode))
    }
}

tryGet(origNode node, key []byte, pos int)方法提供了三个参数，初始节点，hash key，当前hash匹配位置。 didResolve用于判断trie树是否发生变化，根据hashNode去db中获取节点值。 拿到之后需要更新已有的trie，didResolve会发生变化。

Trie的Update和Delete就不分析了。 trie 包中还有其他函数。 不详细解释，先看看主要功能：

databases.go trie数据结构和磁盘数据库之间的写层，方便对trie中的节点进行插入和删除

iterator.go 遍历Trie的key-value迭代器

proof.go Trie树Merkle证明，Prove方法获取指定Key的证明证明，证明证明是从根节点到叶子节点的所有节点的哈希值列表 。 VerifyProof 方法接受根哈希值和证明证书和密钥以验证密钥是否存在。

security_trie.go 加密的trie实现