账号状态存储在MPT中的应用

关于MPT，这篇文章已经有很详细的介绍了：https://ethfans.org/posts/merkle-patricia-tree-in-detail 。所以本文不聊MPT的相关内容，只聊聊账号在MPT中是怎么存储的。

World State，也就是世界状态，主要就是用来存储账户的状态的。可以根据块号查询某个账户的历史信息（余额，交易数），也可以通过最新块号查询很久都没有交易的账户信息都是通过这个世界状态来实现的。而这些状态的存储都是通过MPT来实现的。为了方便理解，会先说代码逻辑，然后以示例的方式展示账户的存储和更新过程。

既然state中存储了账户的信息，又能根据块来查询账户信息，那是如何根据块号来定位到账户信息的呢。来看接口

func (s *PublicBlockChainAPI) GetBalance(ctx context.Context, address common.Address, blockNr rpc.BlockNumber) (*big.Int, error) {state, _, err := s.b.StateAndHeaderByNumber(ctx, blockNr)if state == nil || err != nil {return nil, err}b := state.GetBalance(address)return b, state.Error()
}

查询账户余额的时候，需要先根据块号来找到对应的state。块头中有一个Root字段，存储的就是当前块在MTP中的索引，所以这里的state是通过header.root来定位的。

func (b *HpbApiBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error) {if blockNr == rpc.PendingBlockNumber {block, state := b.hpb.miner.Pending()return state, block.Header(), nil}// Otherwise resolve the block number and return its stateheader, err := b.HeaderByNumber(ctx, blockNr)if header == nil || err != nil {return nil, nil, err}stateDb, err := b.hpb.BlockChain().StateAt(header.Root)return stateDb, header, err
}

现在知道了Root是入口，那接下来看看这个Root是如何生成的，以及账户信息是如何与Root关联起来的。为了简化模型，咱们从零开始，假设现在一个链，刚做了初始化，块0的Root为0x18f74973，此时该root下没有关联账户。然后设置了挖矿账户为0x000001。开启挖矿，第一个块挖出来了。现在要对挖矿账户进行奖励了。

现在要解决的问题是

1、什么时候对挖矿账户进行奖励？

2、如何对挖矿账户添加奖励呢？

3、如何将账户与Root关联起来的。

第一个问题简单，既然是挖矿，当然是当块要写入到链中的时候进行奖励。这个不用细说。看第二个问题，如何如何对挖矿账户添加奖励呢？看代码，添加奖励是通过statedb的AddBalance接口进行的。如果statedb中能查到账户对应的state_object，就使用查到的，如果查不到就新建一个对应的object。

func (self *StateDB) AddBalance(addr common.Address, amount *big.Int) {//log.Error("AddBalance", "addr", addr)stateObject := self.GetOrNewStateObject(addr)if stateObject != nil {stateObject.AddBalance(amount)}
}
func (self *StateDB) GetOrNewStateObject(addr common.Address) *stateObject {stateObject := self.getStateObject(addr)if stateObject == nil || stateObject.deleted {stateObject, _ = self.createObject(addr)}return stateObject
}

看一下statedb与stateobject的数据结构

type StateDB struct {db   Databasetrie TriestateObjects      map[common.Address]*stateObjectstateObjectsDirty map[common.Address]struct{}dbErr errorrefund *big.Intthash, bhash common.HashtxIndex      intlogs         map[common.Hash][]*types.LoglogSize      uintpreimages map[common.Hash][]bytejournal        journalvalidRevisions []revisionnextRevisionId intlock sync.Mutex
}
type Account struct {Nonce    uint64Balance  *big.IntRoot     common.Hash // merkle root of the storage trieCodeHash []byte
}
type stateObject struct {address  common.AddressaddrHash common.Hash // hash of hpb address of the accountdata     Accountdb       *StateDBdbErr errortrie Trie // storage trie, which becomes non-nil on first accesscode Code // contract bytecode, which gets set when code is loadedcachedStorage Storage // Storage entry cache to avoid duplicate readsdirtyStorage  Storage // Storage entries that need to be flushed to diskdirtyCode bool // true if the code was updatedsuicided  booltouched   booldeleted   boolonDirty   func(addr common.Address) // Callback method to mark a state object newly dirty
}

这里我们重点关注的是data account，account中存储了账号的Balance，即余额。至此可知，statedb中有了挖矿账户对应的stateobject，并存储在stateObjects中，对于新账户是在createObject时放入stateObjects中的。

现在看第3个问题，如何将上面的stateObjects更新到MPT中。在更新奖励的时候，会统一将statedb中的数据进行提交，代码如下：通过IntermediateRoot–>Finalise–>updateStateObject,然后返回s.trie.Hash()，即Root。

func (s *StateDB) Finalise(deleteEmptyObjects bool) {for addr := range s.stateObjectsDirty {stateObject := s.stateObjects[addr]if stateObject.suicided || (deleteEmptyObjects && stateObject.empty()) {s.deleteStateObject(stateObject)} else {stateObject.updateRoot(s.db)s.updateStateObject(stateObject)}}s.clearJournalAndRefund()}func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {s.Finalise(deleteEmptyObjects)return s.trie.Hash()
}

现在看看updateStateObject是如何处理的。

func (self *StateDB) updateStateObject(stateObject *stateObject) {addr := stateObject.Address()data, err := rlp.EncodeToBytes(stateObject)if err != nil {panic(fmt.Errorf("can't encode object at %x: %v", addr[:], err))}self.trie.TryUpdate(addr[:], data)
}

通过trie.TryUpdate将addr与encode后的stateobject更新到trie中。看下TryUpdate的细节。代码如下

func (t *Trie) TryUpdate(key, value []byte) error {k := keybytesToHex(key)if len(value) != 0 {log.Warn("zzz TryUpdate", "key", key, "v", value, "k", k)_, n, err := t.insert(t.root, nil, k, valueNode(value))if err != nil {return err}t.root = n} else {_, n, err := t.delete(t.root, nil, k)if err != nil {return err}t.root = n}return nil
}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}return true, value, nil}switch n := n.(type) {case *shortNode:matchlen := prefixLen(key, n.Key) //找到key与n.key前几位重合的长度if matchlen == len(n.Key) { //如果key包含了n.Keydirty, 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}// Otherwise branch out at the index where they differ.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}// Replace this shortNode with the branch if it occurs at index 0.if matchlen == 0 {return true, branch, nil}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]] = nnreturn true, n, nilcase nil:return true, &shortNode{key, value, t.newFlag()}, nilcase hashNode: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, nildefault:panic(fmt.Sprintf("%T: invalid node: %v", n, n))
}

}
TryUpdate的逻辑是判断value的长度，如果大于0就是插入操作，否则就进行删除。这里重点关注插入操作。在插入操作中涉及到4中node类型。fullnode,shortnode,hashnode,valuenode。看下定义

type (fullNode struct {Children [17]node flags    nodeFlag}shortNode struct {Key   []byte Val   node   flags nodeFlag}hashNode  []bytevalueNode []byte
)

fullnode虽然定义了17个children，但是只用了前16个，node可以是4种中任意类型。MPT是默克尔树与前缀树的合并，就是通过这4中node来实现的。继续看下insert的操作。

因为现在只有一个账户需要更新，而且在insert时传入的node类型为nil(因为块0的root没有关联账号)，所以会走到case nil分支，生成一个shortnode，这时，root下对应的就是挖矿账号0x000001的信息。假设块1的Root为0x9eae6bdf，那此时的关联关系为0x9eae6bdf：[shortnode{0x000001:data}],data是经过编码的stateobject，包含了account信息，另外，这里先忽略nodeflag，之后用到时再说。

需要补充说明的是，块1的Root是在账号更新完毕后才生成的，因此演化过程是块0的Root对应一个空：block(0).Root:[] ，中间过程：block(0).Root:[shortnode{0x000001:data}]，最后根据shortnode{0x000001:data}计算出新的block(1).Root，并更新成block(1).Root，然后写入数据库。如此一来，查询余额的时候就可以根据块号1，找到block(1).Root,然后找到shortnode{0x000001:data}，然后根据地址0x000001找到data，从data中解析出余额。

假如继续挖矿块2，矿工0x000002挖到了矿，根据上面的insert代码，对0x000002的账户更新过程变成了如下

1、块1的root下关联了矿工0x000001的账户，而且是shortnode类型，因此走到shortnode分支。

此时状态为：block(1).Root:[shortnode{0x000001:data}]

2、账户0x000001与账户0x000002有着共同的前缀0x00000,长度为5,与账户0x000001的账户长度6不相等，因此不是同一个账户。

3、生成一个fullnode，将账号0x000001写入到fullnode中。

此时fullnode为 fullnode{ 0:nil, 1:shortnode{0x1:data},2:nil,3:nil…15:nil,16:nil }

4、将账号0x000002写入到fullnode中。

此时fullnode为fullnode{ 0:nil, 1:shortnode{0x1:data},2:shortnode{0x2:data},3:nil…15:nil,16:nil }

5、因为matchlen不等于0，所以将shortnode更新后返回。

此时状态为：block(1).Root:[ shortnode{0x00000:fullnode{0:nil, 1:shortnode{0x1:data},2:shortnode{0x2:data},3:nil…15:nil,16:nil}} ]

6、然后根据账户信息重新计算出新块的Root，并将block(2).Root:[ shortnode{0x00000:fullnode{0:nil, 1:shortnode{0x1:data},2:shortnode{0x2:data},3:nil…15:nil,16:nil}} ]写入到数据库中。

假设上面的账户2与账户1没有重叠的前缀，就会返回一个fullnode。有兴趣的可以自己写一写。node内部的shortnode可能会合并为一个fullnode，也可能会包含一个fullnode。Root对应的值以hashnode的方式存储，上面的data就是以valuenode的方式存储。

这里就会产生一个问题了，根据以上内容可知，随着账户的增多，Root对应的node也会越来越大，那每次更新到数据库的值也就越来越大。如何避免呢？

数据量的增长是无法避免的，但可以缓解，这是就需要用到nodeflag了。看看nodeflag是什么：

type nodeFlag struct {hash  hashNode // cached hash of the node (may be nil)gen   uint16   // cache generation counterdirty bool
}
func (t *Trie) newFlag() nodeFlag {return nodeFlag{dirty: true, gen: t.cachegen}
}

nodeflag中重点关注下gen和dirty，gen可以认为是node生成的时间，dirty表示node是否被修改。还记得前文介绍的s.trie.Hash()，看看Hash()都做了什么，代码如下

func (t *Trie) Hash() common.Hash {hash, cached, _ := t.hashRoot(nil)t.root = cachedlog.Warn("zzz Hash", "hashnode", common.BytesToHash(hash.(hashNode)))return common.BytesToHash(hash.(hashNode))
}
func (t *Trie) hashRoot(db DatabaseWriter) (node, node, error) {if t.root == nil {log.Warn("zzz hashRoot", "emptyRoot", hashNode(emptyRoot.Bytes()))return hashNode(emptyRoot.Bytes()), nil, nil}h := newHasher(t.cachegen, t.cachelimit)defer returnHasherToPool(h)log.Warn("zzz hashRoot", "Root", t.root)return h.hash(t.root, db, true)
}
func (h *hasher) hash(n node, db DatabaseWriter, force bool) (node, node, error) {// If we're not storing the node, just hashing, use available cached dataif hash, dirty := n.cache(); hash != nil {if db == nil {return hash, n, nil}if n.canUnload(h.cachegen, h.cachelimit) {.cacheUnloadCounter.Inc(1)return hash, hash, nil}if !dirty {return hash, n, nil}}collapsed, cached, err := h.hashChildren(n, db)if err != nil {return hashNode{}, n, err}hashed, err := h.store(collapsed, db, force)if err != nil {return hashNode{}, n, err}cachedHash, _ := hashed.(hashNode)switch cn := cached.(type) {case *shortNode:cn.flags.hash = cachedHashif db != nil {cn.flags.dirty = false}case *fullNode:cn.flags.hash = cachedHashif db != nil {cn.flags.dirty = false}}return hashed, cached, nil
}

Hash()的作用是将root对应的值以hashnode的形式返回，其中调用了hashRoot函数，该函数中又主要是调用hash函数，注意hash函数中有个canUnload函数调用，如果canUnload成立，则直接返回node ，在往下看，通过store函数将node进行存储并返回存储node的hash值。canUnload的判断标准看代码，cachelimit默认是120。

func (n *nodeFlag) canUnload(cachegen, cachelimit uint16) bool {return !n.dirty && cachegen-n.gen >= cachelimit
}

而cachegen是在每次执行committo函数调用时加1，n.gen是创建node时设置的，代码如下：

func (t *Trie) CommitTo(db DatabaseWriter) (root common.Hash, err error) {hash, cached, err := t.hashRoot(db)if err != nil {return (common.Hash{}), err}t.root = cachedt.cachegen++return common.BytesToHash(hash.(hashNode)), nil
}

也就是每经过120次调用后，没有修改的node将被hash代替，这样就降低了存储的量。

get,delete操作都是根据key一层层往下找，找到取出或者删除，搞懂了insert,这两个的逻辑上就不复杂了。

到这里，账户在MPT中的应用就结束了。

账号状态存储在MPT中的应用相关推荐

android 读取内部存储文件格式,Android中的数据储存之文件存储
当我们在使用各种程序时,其实际上是在和各种数据打交道,当我们聊QQ,刷微博,看新闻,其实都是在和里面的数据交互例如在聊天时发出的消息,以及在登录时输入的账号密码,其实都是瞬时数据,那什么是瞬时数据呢 ...
小猪的Python学习之旅 —— 20.抓取Gank.io所有数据存储到MySQL中
小猪的Python学习之旅 -- 20.抓取Gank.io所有数据存储到MySQL中标签:Python 一句话概括本文: 内容较多,建议先mark后看,讲解了一波MySQL安装,基本操作,语法速成, ...
Rook存储：Kubernetes中最优秀的存储
本文讲的是Rook存储:Kubernetes中最优秀的存储[编者的话]Rook存储集群,其实是在著名的分布式存储系统Ceph的一个封装,以Kubernetes Application的方式运行了监控. ...
mysql连接卡死，很多线程sleep状态，导致CPU中mysqld占用率极高（问题原因还待考证）...
关闭所有 .................................. .连接: ##把全部的MySQL连接kill掉 for i in $(mysql -uroot -p123456 -Bs ...
md5后得到的32位字符串存储到mysql中太占空间了_好看！快收藏：非常完整的 MySQL 规范...
一.数据库命令规范所有数据库对象名称必须使用小写字母并用下划线分割所有数据库对象名称禁止使用mysql保留关键字(如果表名中包含关键字查询时,需要将其用单引号括起来) 数据库对象的命名要能做到见名 ...
mysql sleep 5908_mysql连接卡死，很多线程sleep状态，导致CPU中mysqld占用率极高（问题原因还待考证）...
mysql> show processlist; +-–+-----+------–+ | Id | User | Host | db | Command | Time| State | Inf ...
Flink之状态之状态存储 state backends
流计算中可能有各种方式来保存状态: 窗口操作使用了KV操作的函数继承了CheckpointedFunction的函数当开始做checkpointing的时候,状态会被持久化到checkpoin ...
从文件中读取并进行树的存储_数据库中的面试题你能接几招
(附答案,不带答案的面试题都是耍流氓) 1. 事务的特性 ACID: 原子性, 一致性, 隔离性, 持久性 2. innodb如何结果幻读在不可重复读的隔离级别下使用间隙锁 3. 什么是间隙锁 In ...
md5后得到的32位字符串存储到mysql中太占空间了_面试官：你对MySQL高性能优化有什么规范建议？...
推荐阅读:吊打面试官!MySQL灵魂100问,你能答出多少? 文章篇幅较长,建议先收藏再找个合适的时间阅读数据库命令规范所有数据库对象名称必须使用小写字母并用下划线分割所有数据库对象名称禁止使用 ...

账号状态存储在MPT中的应用

账号状态存储在MPT中的应用相关推荐

最新文章

热门文章