state

在以太坊中，交易驱动state的更新，而state的更新影响它底层的trie。如下，是一个交易对state的一个更改的示意图。

由图可见，将账号x中的金额从100改为110之后，影响了其中的4个节点。即如果改变一个账户的数据，沿着数据往上的父节点的数据都将发生改变。如果每次产生一个区块都将整颗状态树的数据保存到硬盘中，那么硬盘将保存了许多的不需要的历史数据。截至到2021/11/29为止。采用Ethereum Full Node Sync (Archive) Chart 即归档节点保存所有历史数据的模式，数据量已达到9116Gb，而采用Ethereum Full Node Sync (Default) Chart即全节点保存最新数据的模式，数据量为1089Gb。前者大概是后者的9倍。

修剪

先将修剪相关的代码放过来，为了理清代码逻辑，只保留相关代码：

func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool) (status WriteStatus, err error) {
    ......

    // Commit all cached state changes into underlying memory database.
    root, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
    if err != nil {
        return NonStatTy, err
    }
    triedb := bc.stateCache.TrieDB()

    // If we're running an archive node, always flush
    if bc.cacheConfig.TrieDirtyDisabled {
        if err := triedb.Commit(root, false, nil); err != nil {
            return NonStatTy, err
        }
    } else {
        // Full but not archive node, do proper garbage collection
        triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive
        bc.triegc.Push(root, -int64(block.NumberU64()))

        if current := block.NumberU64(); current > TriesInMemory {
            // If we exceeded our memory allowance, flush matured singleton nodes to disk
            var (
                nodes, imgs = triedb.Size()
                limit       = common.StorageSize(bc.cacheConfig.TrieDirtyLimit) * 1024 * 1024
            )
            if nodes > limit || imgs > 4*1024*1024 {
                triedb.Cap(limit - ethdb.IdealBatchSize)
            }
            // Find the next state trie we need to commit
            chosen := current - TriesInMemory

            // If we exceeded out time allowance, flush an entire trie to disk
            if bc.gcproc > bc.cacheConfig.TrieTimeLimit {
                // If the header is missing (canonical chain behind), we're reorging a low
                // diff sidechain. Suspend committing until this operation is completed.
                header := bc.GetHeaderByNumber(chosen)
                if header == nil {
                    log.Warn("Reorg in progress, trie commit postponed", "number", chosen)
                } else {
                    // If we're exceeding limits but haven't reached a large enough memory gap,
                    // warn the user that the system is becoming unstable.
                    if chosen < lastWrite+TriesInMemory && bc.gcproc >= 2*bc.cacheConfig.TrieTimeLimit {
                        log.Info("State in memory for too long, committing", "time", bc.gcproc, "allowance", bc.cacheConfig.TrieTimeLimit, "optimum", float64(chosen-lastWrite)/TriesInMemory)
                    }
                    // Flush an entire trie and restart the counters
                    triedb.Commit(header.Root, true, nil)
                    lastWrite = chosen
                    bc.gcproc = 0
                }
            }
            // Garbage collect anything below our required write retention
            for !bc.triegc.Empty() {
                root, number := bc.triegc.Pop()
                if uint64(-number) > chosen {
                    bc.triegc.Push(root, number)
                    break
                }
                triedb.Dereference(root.(common.Hash))
            }
        }
    }
    ......
}

以代码的顺序，逐步进行分析：

代码中的 state.Commit(...) 并没有将数据直接写到leveldb中，只是写入到了trie对应的内存database中，即triedb := bc.stateCache.TrieDB()拿到的 triedb，后面所有的修剪，其实就是对triedb进行修剪。

拿到内存数据库之后，紧接着做一个判断bc.cacheConfig.TrieDirtyDisabled，如果是归档节点，那么直接调用triedb的提交函数，将triedb中所有节点数据提交到leveldb中。

如果不是归档节点，那么调用 triedb.Reference(root, common.Hash{}) 将当前的triedb里面当前的trie进行引用。这样确保go的垃圾回收机制不会将这颗trie进行回收。

进行引用完之后，将这棵树对应的root与区块高度推送到 bc.triegc.Push(root, -int64(block.NumberU64())) 优先队列里面去。triegc的Pop操作是一个将当前队列中优先级值最大的数据弹出队列并返回的操作。对于trie的清理，应该是从比较旧的区块进行清理，所以triege在进行Push操作的时候，将区块高度取反变为负数存进去。

紧接着进行修剪判断，只有当当前区块高度大于 TriesInMemory(当前代码中定义为128) 的时候。 而且，后面对triedb里面的trie进行解引用triedb.Dereference(root.(common.Hash))的时候，只对当前区块减去128个区块之前的区块所对应的trie进行解引用。为什么要这样做呢？因为以太坊存在分叉，系统不知道什么时候需要进行回滚。所以不能简单的当出完区块高度为 N 的区块高度之后，就立即对区块高度为 N-1 的区块进行解引用。

进入修剪逻辑之后，拿到triedb里面的两个数据值nodes, imgs，前者nodes对应的是triedb里面所有trie所占用的内存，后者imgs对应的是SecureTrie中的key-value对所占用的内存。当所有trie占用内存超过 limit = common.StorageSize(bc.cacheConfig.TrieDirtyLimit) * 1024 * 1024 (当前代码定义为256M)，或者imgs占用超过4M (一条key-value对占用内存 20 + 32 即写入大概8万条数据)的时候。系统会强制将triedb里面比较旧的数据直接写入到leveldb，将triedb占用的内存减少到限制的 limit 之内。注意，此时的有可能不需要的旧数据（即后续有可能被剪掉的数据）数据写入到leveldb中。

紧接着又进行一个事件的判断 if bc.gcproc > bc.cacheConfig.TrieTimeLimit。其中bc.cacheConfig.TrieTimeLimit默认为5分钟。如果符合这个条件，则直接对最新的trie进行triedb.Commit(header.Root, true, nil)写入到leveldb不进行任何修改。

先找到gcproc相关代码，如下所示：

func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, error) {
    it := newInsertIterator(chain, results, bc.validator)
    ......
    for ; block != nil && err == nil || err == ErrKnownBlock; block, err = it.next() {
        proctime := time.Since(start)
        ......
        switch status {
        case CanonStatTy:
            ......
            // Only count canonical blocks for GC processing time
            bc.gcproc += proctime
            ......
        }
    }
    ......
}

这是一段区块同步的代码，也就是说。如果我同步区块处理state的所累积的时间超过了5分钟，那么我不做任何修剪进行直接写leveldb。

最后就是进行修剪也就是解引用了，从队里里面将最老的区块对应的trie依次弹出进行解引用。当解引用操作拿出来的区块高度大于当前区块减去128个区块的时候，解应用操作结束。

总结

• 当triedb里面的的节点或者安全trie的key-value对占用的内存超过设定的阈值时，triedb会将内存里面的数据写入leveldb并从内存中释放掉写入leveldb中的数据。
• 当更新state累计的时间超过一定的时间时，直接将最新的trie写入leveldb中。
• 修剪当前区块减去128个区块对应的trie，将没有使用到的节点从内存中释放。