forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-24 21:56:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core/state: rewrite the state size tracker

Browse source
This commit is contained in:
Gary Rong 2025-08-15 19:55:51 +08:00
parent b62ea6074d
commit 0861e6dffe
10 changed files with 582 additions and 562 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
core/blockchain.go
View file

@ -196,6 +196,9 @@ type BlockChainConfig struct {
// If the value is zero, all transactions of the entire chain will be indexed.
// If the value is -1, indexing is disabled.
TxLookupLimit int64
// EnableStateSizeTracking indicates whether the state size tracking is enabled.
EnableStateSizeTracking bool
}
// DefaultConfig returns the default config.
@ -333,8 +336,7 @@ type BlockChain struct {
prefetcher Prefetcher
processor Processor // Block transaction processor interface
logger *tracing.Hooks
stateSizeGen *state.StateSizeGenerator // State size tracking
stateSizer *state.SizeTracker // State size tracking
lastForkReadyAlert time.Time // Last time there was a fork readiness print out
}
@ -530,9 +532,10 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine,
}
// Start state size tracker
bc.stateSizeGen = state.NewStateSizeGenerator(bc.statedb.DiskDB(), bc.triedb, head.Root)
log.Info("Started state size generator", "root", head.Root)
if bc.cfg.EnableStateSizeTracking {
bc.stateSizer = state.NewSizeTracker(bc.db, bc.triedb)
bc.stateSizer.Start()
}
return bc, nil
}
@ -1259,12 +1262,10 @@ func (bc *BlockChain) stopWithoutSaving() {
// Signal shutdown to all goroutines.
bc.InterruptInsert(true)
// Stop state size generator if running
if bc.stateSizeGen != nil {
bc.stateSizeGen.Stop()
log.Info("Stopped state size generator")
// Stop state size tracker
if bc.stateSizer != nil {
bc.stateSizer.Stop()
}
// Now wait for all chain modifications to end and persistent goroutines to exit.
//
// Note: Close waits for the mutex to become available, i.e. any running chain
@ -1604,10 +1605,9 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
if err != nil {
return err
}
// Track state size changes if generator is running
if bc.stateSizeGen != nil && stateUpdate != nil {
bc.stateSizeGen.Track(stateUpdate)
// Emit the state update to the state sizestats if it's active
if bc.stateSizer != nil {
bc.stateSizer.Notify(stateUpdate)
}
// If node is running in path mode, skip explicit gc operation
// which is unnecessary in this mode.

10
core/state/metrics.go
View file

@ -29,14 +29,4 @@ var (
storageTriesUpdatedMeter = metrics.NewRegisteredMeter("state/update/storagenodes", nil)
accountTrieDeletedMeter = metrics.NewRegisteredMeter("state/delete/accountnodes", nil)
storageTriesDeletedMeter = metrics.NewRegisteredMeter("state/delete/storagenodes", nil)
// State size metrics
accountCountGauge = metrics.NewRegisteredGauge("state/account/count", nil)
accountBytesGauge = metrics.NewRegisteredGauge("state/account/bytes", nil)
storageCountGauge = metrics.NewRegisteredGauge("state/storage/count", nil)
storageBytesGauge = metrics.NewRegisteredGauge("state/storage/bytes", nil)
trienodeCountGauge = metrics.NewRegisteredGauge("state/trienode/count", nil)
trienodeBytesGauge = metrics.NewRegisteredGauge("state/trienode/bytes", nil)
contractCountGauge = metrics.NewRegisteredGauge("state/contract/count", nil)
contractBytesGauge = metrics.NewRegisteredGauge("state/contract/bytes", nil)
)

547
core/state/state_sizer.go Normal file
View file

@ -0,0 +1,547 @@
// Copyright 2025 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package state
import (
"container/heap"
"errors"
"fmt"
"maps"
"slices"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/triedb"
"golang.org/x/sync/errgroup"
)
const (
statEvictThreshold = 128 // the depth of statistic to be preserved
)
// Metrics for uploading the state statistics.
var (
accountsGauge = metrics.NewRegisteredGauge("state/size/account/count", nil)
accountBytesGauge = metrics.NewRegisteredGauge("state/size/account/bytes", nil)
storagesGauge = metrics.NewRegisteredGauge("state/size/storage/count", nil)
storageBytesGauge = metrics.NewRegisteredGauge("state/size/storage/bytes", nil)
accountTrienodesGauge = metrics.NewRegisteredGauge("state/size/trienode/account/count", nil)
accountTrienodeBytesGauge = metrics.NewRegisteredGauge("state/size/trienode/account/bytes", nil)
storageTrienodesGauge = metrics.NewRegisteredGauge("state/size/trienode/storage/count", nil)
storageTrienodeBytesGauge = metrics.NewRegisteredGauge("state/size/trienode/storage/bytes", nil)
contractCodesGauge = metrics.NewRegisteredGauge("state/size/contractcode/count", nil)
contractCodeBytesGauge = metrics.NewRegisteredGauge("state/size/contractcode/bytes", nil)
)
// Database key scheme for states.
var (
accountKeySize = int64(len(rawdb.SnapshotAccountPrefix) + common.HashLength)
storageKeySize = int64(len(rawdb.SnapshotStoragePrefix) + common.HashLength + common.HashLength)
accountTrienodePrefixSize = int64(len(rawdb.TrieNodeAccountPrefix))
storageTrienodePrefixSize = int64(len(rawdb.TrieNodeStoragePrefix) + common.HashLength)
codeKeySize = int64(len(rawdb.CodePrefix) + common.HashLength)
)
// SizeStats represents either the current state size statistics or the size
// differences resulting from a state transition.
type SizeStats struct {
StateRoot common.Hash // State root hash at the time of measurement
BlockNumber uint64 // Associated block number at the time of measurement
Accounts int64 // Total number of accounts in the state
AccountBytes int64 // Total storage size used by all account data (in bytes)
Storages int64 // Total number of storage slots across all accounts
StorageBytes int64 // Total storage size used by all storage slot data (in bytes)
AccountTrienodes int64 // Total number of account trie nodes in the state
AccountTrienodeBytes int64 // Total storage size occupied by account trie nodes (in bytes)
StorageTrienodes int64 // Total number of storage trie nodes in the state
StorageTrienodeBytes int64 // Total storage size occupied by storage trie nodes (in bytes)
ContractCodes int64 // Total number of contract codes in the state
ContractCodeBytes int64 // Total size of all contract code (in bytes)
}
// add applies the given state diffs and produces a new version of the statistics.
func (s SizeStats) add(diff SizeStats) SizeStats {
var combo SizeStats
combo.StateRoot = diff.StateRoot
combo.BlockNumber = diff.BlockNumber
combo.Accounts += diff.Accounts
combo.AccountBytes += diff.AccountBytes
combo.Storages += diff.Storages
combo.StorageBytes += diff.StorageBytes
combo.AccountTrienodes += diff.AccountTrienodes
combo.AccountTrienodeBytes += diff.AccountTrienodeBytes
combo.StorageTrienodes += diff.StorageTrienodes
combo.StorageTrienodeBytes += diff.StorageTrienodeBytes
combo.ContractCodes += diff.ContractCodes
combo.ContractCodeBytes += diff.ContractCodeBytes
return combo
}
// calSizeStats measures the state size changes of the provided state update.
func calSizeStats(update *stateUpdate) (SizeStats, error) {
stats := SizeStats{
BlockNumber: update.blockNumber,
StateRoot: update.root,
}
// Measure the account changes
for addr, oldValue := range update.accountsOrigin {
addrHash := crypto.Keccak256Hash(addr.Bytes())
newValue, exists := update.accounts[addrHash]
if !exists {
return SizeStats{}, fmt.Errorf("account %x not found", addr)
}
oldLen, newLen := len(oldValue), len(newValue)
switch {
case oldLen > 0 && newLen == 0:
// Account deletion
stats.Accounts -= 1
stats.AccountBytes -= accountKeySize + int64(oldLen)
case oldLen == 0 && newLen > 0:
// Account creation
stats.Accounts += 1
stats.AccountBytes += accountKeySize + int64(newLen)
default:
// Account update
stats.AccountBytes += int64(newLen - oldLen)
}
}
// Measure storage changes
for addr, slots := range update.storagesOrigin {
addrHash := crypto.Keccak256Hash(addr.Bytes())
subset, exists := update.storages[addrHash]
if !exists {
return SizeStats{}, fmt.Errorf("storage %x not found", addr)
}
for key, oldValue := range slots {
var (
exists bool
newValue []byte
)
if update.rawStorageKey {
newValue, exists = subset[crypto.Keccak256Hash(key.Bytes())]
} else {
newValue, exists = subset[key]
}
if !exists {
return SizeStats{}, fmt.Errorf("storage slot %x-%x not found", addr, key)
}
oldLen, newLen := len(oldValue), len(newValue)
switch {
case oldLen > 0 && newLen == 0:
// Storage deletion
stats.Storages -= 1
stats.StorageBytes -= storageKeySize + int64(oldLen)
case oldLen == 0 && newLen > 0:
// Storage creation
stats.Storages += 1
stats.StorageBytes += storageKeySize + int64(newLen)
default:
// Storage update
stats.StorageBytes += int64(newLen - oldLen)
}
}
}
// Measure trienode changes
for owner, subset := range update.nodes.Sets {
var (
keyPrefix int64
isAccount = owner == (common.Hash{})
)
if isAccount {
keyPrefix = accountTrienodePrefixSize
} else {
keyPrefix = storageTrienodePrefixSize
}
// Iterate over Origins since every modified node has an origin entry
for path, oldNode := range subset.Origins {
newNode, exists := subset.Nodes[path]
if !exists {
return SizeStats{}, fmt.Errorf("node %x-%v not found", owner, path)
}
keySize := keyPrefix + int64(len(path))
switch {
case len(oldNode) > 0 && len(newNode.Blob) == 0:
// Node deletion
if isAccount {
stats.AccountTrienodes -= 1
stats.AccountTrienodeBytes -= keySize + int64(len(oldNode))
} else {
stats.StorageTrienodes -= 1
stats.StorageTrienodeBytes -= keySize + int64(len(oldNode))
}
case len(oldNode) == 0 && len(newNode.Blob) > 0:
// Node creation
if isAccount {
stats.AccountTrienodes += 1
stats.AccountTrienodeBytes += keySize + int64(len(newNode.Blob))
} else {
stats.StorageTrienodes += 1
stats.StorageTrienodeBytes += keySize + int64(len(newNode.Blob))
}
default:
// Node update
if isAccount {
stats.AccountTrienodeBytes += int64(len(newNode.Blob) - len(oldNode))
} else {
stats.StorageTrienodeBytes += int64(len(newNode.Blob) - len(oldNode))
}
}
}
}
// Measure code changes. Note that the reported contract code size may be slightly
// inaccurate due to database deduplication (code is stored by its hash). However,
// this deviation is negligible and acceptable for measurement purposes.
for _, code := range update.codes {
stats.ContractCodes += 1
stats.ContractCodeBytes += codeKeySize + int64(len(code.blob))
}
return stats, nil
}
// SizeTracker handles the state size initialization and tracks of state size metrics.
type SizeTracker struct {
db ethdb.KeyValueStore
triedb *triedb.Database
abort chan struct{}
aborted chan struct{}
updateCh chan *stateUpdate
}
// NewSizeTracker creates a new state size tracker and starts it automatically
func NewSizeTracker(db ethdb.KeyValueStore, triedb *triedb.Database) *SizeTracker {
return &SizeTracker{
db: db,
triedb: triedb,
abort: make(chan struct{}),
aborted: make(chan struct{}),
updateCh: make(chan *stateUpdate),
}
}
func (t *SizeTracker) Start() {
go t.run()
}
func (t *SizeTracker) Stop() {
close(t.abort)
<-t.aborted
}
// sizeStatsHeap is a heap.Interface implementation over statesize statistics for
// retrieving the oldest statistics for eviction.
type sizeStatsHeap []SizeStats
func (h sizeStatsHeap) Len() int { return len(h) }
func (h sizeStatsHeap) Less(i, j int) bool { return h[i].BlockNumber < h[j].BlockNumber }
func (h sizeStatsHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *sizeStatsHeap) Push(x any) {
*h = append(*h, x.(SizeStats))
}
func (h *sizeStatsHeap) Pop() any {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// run performs the state size initialization and handles updates
func (t *SizeTracker) run() {
defer close(t.aborted)
stats, err := t.init() // launch background thread for state size init
if err != nil {
return
}
h := sizeStatsHeap(slices.Collect(maps.Values(stats)))
heap.Init(&h)
for {
select {
case u := <-t.updateCh:
base, found := stats[u.originRoot]
if !found {
continue
}
diff, err := calSizeStats(u)
if err != nil {
continue
}
stat := base.add(diff)
t.upload(stat)
stats[u.root] = stat
heap.Push(&h, stats[u.root])
for u.blockNumber-h[0].BlockNumber > statEvictThreshold {
heap.Pop(&h)
}
case <-t.abort:
return
}
}
}
type buildResult struct {
stat SizeStats
root common.Hash
blockNumber uint64
err error
}
func (t *SizeTracker) init() (map[common.Hash]SizeStats, error) {
// Wait for snapshot completion and then init
ticker := time.NewTicker(10 * time.Second)
defer ticker.Stop()
wait:
for {
select {
case <-ticker.C:
if t.triedb.SnapshotCompleted() {
break wait
}
case <-t.abort:
return nil, errors.New("size tracker closed")
}
}
var (
updates = make(map[common.Hash]*stateUpdate)
children = make(map[common.Hash][]common.Hash)
done = make(chan buildResult)
)
for {
select {
case u := <-t.updateCh:
updates[u.root] = u
children[u.originRoot] = append(children[u.originRoot], u.root)
case <-ticker.C:
root := rawdb.ReadSnapshotRoot(t.db)
if root == (common.Hash{}) {
continue
}
entry, exists := updates[root]
if !exists {
continue
}
if done == nil {
done = make(chan buildResult)
go t.build(entry.root, entry.blockNumber, done)
}
case result := <-done:
if result.err != nil {
return nil, result.err
}
var (
stats = make(map[common.Hash]SizeStats)
apply func(root common.Hash, stats SizeStats) error
)
apply = func(root common.Hash, stat SizeStats) error {
for _, child := range children[root] {
entry, ok := updates[child]
if !ok {
return fmt.Errorf("the state update is not found, %x", child)
}
diff, err := calSizeStats(entry)
if err != nil {
return err
}
stats[child] = stat.add(diff)
if err := apply(child, stats[child]); err != nil {
return err
}
}
return nil
}
if err := apply(result.root, result.stat); err != nil {
return nil, err
}
return stats, nil
case <-t.abort:
return nil, errors.New("size tracker closed")
}
}
}
func (t *SizeTracker) build(root common.Hash, blockNumber uint64, done chan buildResult) {
// Metrics will be directly updated by each goroutine
var (
accounts, accountBytes int64
storages, storageBytes int64
codes, codeBytes int64
accountTrienodes, accountTrienodeBytes int64
storageTrienodes, storageTrienodeBytes int64
group errgroup.Group
)
// Start all table iterations concurrently with direct metric updates
group.Go(func() error {
count, bytes, err := t.iterateTable(t.abort, rawdb.SnapshotAccountPrefix, "account")
if err != nil {
return err
}
accounts, accountBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := t.iterateTable(t.abort, rawdb.SnapshotStoragePrefix, "storage")
if err != nil {
return err
}
storages, storageBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := t.iterateTable(t.abort, rawdb.TrieNodeAccountPrefix, "accountnode")
if err != nil {
return err
}
accountTrienodes, accountTrienodeBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := t.iterateTable(t.abort, rawdb.TrieNodeStoragePrefix, "storagenode")
if err != nil {
return err
}
storageTrienodes, storageTrienodeBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := t.iterateTable(t.abort, rawdb.CodePrefix, "contractcode")
if err != nil {
return err
}
codes, codeBytes = count, bytes
return nil
})
// Wait for all goroutines to complete
if err := group.Wait(); err != nil {
done <- buildResult{err: err}
} else {
stat := SizeStats{
StateRoot: root,
BlockNumber: blockNumber,
Accounts: accounts,
AccountBytes: accountBytes,
Storages: storages,
StorageBytes: storageBytes,
AccountTrienodes: accountTrienodes,
AccountTrienodeBytes: accountTrienodeBytes,
StorageTrienodes: storageTrienodes,
StorageTrienodeBytes: storageTrienodeBytes,
ContractCodes: codes,
ContractCodeBytes: codeBytes,
}
done <- buildResult{
root: root,
blockNumber: blockNumber,
stat: stat,
}
}
}
// Notify is an async method used to send the state update to the size tracker.
// It ignores empty updates (where no state changes occurred).
// If the channel is full, it drops the update to avoid blocking.
func (t *SizeTracker) Notify(update *stateUpdate) {
if update == nil || update.empty() {
return
}
select {
case t.updateCh <- update:
case <-t.abort:
return
}
}
// iterateTable performs iteration over a specific table and returns the results.
func (t *SizeTracker) iterateTable(closed chan struct{}, prefix []byte, name string) (int64, int64, error) {
var (
start = time.Now()
logged = time.Now()
count, bytes int64
)
iter := t.db.NewIterator(prefix, nil)
defer iter.Release()
log.Info("Iterating state", "category", name)
for iter.Next() {
count++
bytes += int64(len(iter.Key()) + len(iter.Value()))
if time.Since(logged) > time.Second*8 {
logged = time.Now()
select {
case <-closed:
log.Info("State iteration cancelled", "category", name)
return 0, 0, errors.New("size tracker closed")
default:
log.Info("Iterating state", "category", name, "count", count, "size", common.StorageSize(bytes))
}
}
}
// Check for iterator errors
if err := iter.Error(); err != nil {
log.Error("Iterator error", "category", name, "err", err)
return 0, 0, err
}
log.Info("Finished state iteration", "category", name, "count", count, "size", common.StorageSize(bytes), "elapsed", common.PrettyDuration(time.Since(start)))
return count, bytes, nil
}
func (t *SizeTracker) upload(stats SizeStats) {
accountsGauge.Update(stats.Accounts)
accountBytesGauge.Update(stats.AccountBytes)
storagesGauge.Update(stats.Storages)
storageBytesGauge.Update(stats.StorageBytes)
accountTrienodesGauge.Update(stats.AccountTrienodes)
accountTrienodeBytesGauge.Update(stats.AccountTrienodeBytes)
storageTrienodesGauge.Update(stats.StorageTrienodes)
storageTrienodeBytesGauge.Update(stats.StorageTrienodeBytes)
contractCodesGauge.Update(stats.ContractCodes)
contractCodeBytesGauge.Update(stats.ContractCodeBytes)
}

8
core/state/statedb.go
View file

@ -668,7 +668,7 @@ func (s *StateDB) CreateContract(addr common.Address) {
// Copy creates a deep, independent copy of the state.
// Snapshots of the copied state cannot be applied to the copy.
func (s *StateDB) Copy() *StateDB {
// Copy all the basic fields, initialize the memory ones
// Copy all the basic fields, init the memory ones
state := &StateDB{
db: s.db,
reader: s.reader,
@ -1156,7 +1156,7 @@ func (s *StateDB) GetTrie() Trie {
// commit gathers the state mutations accumulated along with the associated
// trie changes, resetting all internal flags with the new state as the base.
func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) {
func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNumber uint64) (*stateUpdate, error) {
// Short circuit in case any database failure occurred earlier.
if s.dbErr != nil {
return nil, fmt.Errorf("commit aborted due to earlier error: %v", s.dbErr)
@ -1308,13 +1308,13 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool) (*stateU
origin := s.originalRoot
s.originalRoot = root
return newStateUpdate(noStorageWiping, origin, root, deletes, updates, nodes), nil
return newStateUpdate(noStorageWiping, origin, root, blockNumber, deletes, updates, nodes), nil
}
// commitAndFlush is a wrapper of commit which also commits the state mutations
// to the configured data stores.
func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) {
ret, err := s.commit(deleteEmptyObjects, noStorageWiping)
ret, err := s.commit(deleteEmptyObjects, noStorageWiping, block)
if err != nil {
return nil, err
}

2
core/state/statedb_test.go
View file

@ -281,7 +281,7 @@ func TestCopyObjectState(t *testing.T) {
cpy := orig.Copy()
for _, op := range cpy.mutations {
if have, want := op.applied, false; have != want {
t.Fatalf("Error in test itself, the 'done' flag should not be set before Commit, have %v want %v", have, want)
t.Fatalf("Error in test itself, the 'aborted' flag should not be set before Commit, have %v want %v", have, want)
}
}
orig.Commit(0, true, false)

519
core/state/statesize.go
View file

@ -1,519 +0,0 @@
// Copyright 2025 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package state
import (
"context"
"encoding/json"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/triedb"
"golang.org/x/sync/errgroup"
)
var (
accountSnapKeySize = int64(len(rawdb.SnapshotAccountPrefix) + common.HashLength)
storageSnapKeySize = int64(len(rawdb.SnapshotStoragePrefix) + common.HashLength + common.HashLength)
accountTrieKeyPrefixSize = int64(len(rawdb.TrieNodeAccountPrefix))
storageTrieKeyPrefixSize = int64(len(rawdb.TrieNodeStoragePrefix) + common.HashLength)
codeKeySize = int64(len(rawdb.CodePrefix) + common.HashLength)
)
// stateSizeMetrics represents the current state size statistics
type stateSizeMetrics struct {
Root common.Hash // Root hash of the state trie
AccountCount int64
AccountBytes int64
StorageCount int64
StorageBytes int64
TrieNodeCount int64
TrieNodeBytes int64
ContractCount int64
ContractBytes int64
}
// StateSizeGenerator handles the initialization and tracking of state size metrics
type StateSizeGenerator struct {
db ethdb.KeyValueStore
triedb *triedb.Database
abort chan struct{}
done chan struct{}
updateChan chan *stateUpdate // Async message channel for updates
metrics *stateSizeMetrics
buffered *stateSizeMetrics
}
// NewStateSizeGenerator creates a new state size generator and starts it automatically
func NewStateSizeGenerator(db ethdb.KeyValueStore, triedb *triedb.Database, root common.Hash) *StateSizeGenerator {
g := &StateSizeGenerator{
db: db,
triedb: triedb,
abort: make(chan struct{}),
done: make(chan struct{}),
updateChan: make(chan *stateUpdate, 1000), // Buffered channel for updates
metrics: &stateSizeMetrics{Root: root},
buffered: &stateSizeMetrics{Root: root},
}
// Start the generator automatically
go g.generate()
return g
}
// Stop terminates the background generation and persists the metrics.
func (g *StateSizeGenerator) Stop() {
close(g.abort)
<-g.done
g.persistMetrics()
}
// generate performs the state size initialization and handles updates
func (g *StateSizeGenerator) generate() {
defer close(g.done)
var inited bool
// Check if we already have existing metrics
if g.hasExistingMetrics() {
log.Info("State size metrics already initialized")
inited = true
}
initDone := g.initialize()
for {
select {
case update := <-g.updateChan:
g.handleUpdate(update, inited)
case <-g.abort:
log.Info("State size generation aborted")
// Wait for initialization to complete with timeout
if initDone != nil {
select {
case <-initDone:
log.Debug("Initialization completed before abort")
case <-time.After(5 * time.Second):
log.Warn("Initialization did not finish in time during abort")
}
}
return
case <-initDone:
// Initialization completed, merge buffered metrics
if g.buffered != nil {
log.Info("Merging buffered metrics into main metrics")
g.metrics.Root = g.buffered.Root
g.metrics.AccountCount += g.buffered.AccountCount
g.metrics.AccountBytes += g.buffered.AccountBytes
g.metrics.StorageCount += g.buffered.StorageCount
g.metrics.StorageBytes += g.buffered.StorageBytes
g.metrics.TrieNodeCount += g.buffered.TrieNodeCount
g.metrics.TrieNodeBytes += g.buffered.TrieNodeBytes
g.metrics.ContractCount += g.buffered.ContractCount
g.metrics.ContractBytes += g.buffered.ContractBytes
g.buffered = nil
}
inited = true
initDone = nil // Clear the channel to prevent future selects
}
}
}
// initialize starts the initialization process if not already initialized
func (g *StateSizeGenerator) initialize() chan struct{} {
done := make(chan struct{})
// Wait for snapshot completion and then initialize
go func() {
defer close(done)
LOOP:
// Wait for snapshot generator to complete first
for {
root, done := g.triedb.SnapshotCompleted()
if done {
g.metrics.Root = root
g.buffered.Root = root
break LOOP
}
select {
case <-g.abort:
log.Info("State size initialization aborted during snapshot wait")
return
case <-time.After(10 * time.Second):
// Continue checking for snapshot completion
}
}
// Start actual initialization
start := time.Now()
log.Info("Starting state size initialization")
if err := g.initializeMetrics(); err != nil {
log.Error("Failed to initialize state size metrics", "err", err)
return
}
log.Info("Completed state size initialization", "elapsed", time.Since(start))
}()
return done
}
// handleUpdate processes a single update with proper root continuity checking
func (g *StateSizeGenerator) handleUpdate(update *stateUpdate, initialized bool) {
diff := g.calculateUpdateDiff(update)
var targetMetrics *stateSizeMetrics
if initialized {
targetMetrics = g.metrics
} else {
targetMetrics = g.buffered
}
// Check root continuity - the update should build on our current state
if targetMetrics.Root != (common.Hash{}) && targetMetrics.Root != update.originRoot {
log.Warn("State update root discontinuity detected", "current", targetMetrics.Root, "updateOrigin", update.originRoot, "updateNew", update.root)
}
// Update to the new state root
targetMetrics.Root = update.root
targetMetrics.AccountCount += diff.AccountCount
targetMetrics.AccountBytes += diff.AccountBytes
targetMetrics.StorageCount += diff.StorageCount
targetMetrics.StorageBytes += diff.StorageBytes
targetMetrics.TrieNodeCount += diff.TrieNodeCount
targetMetrics.TrieNodeBytes += diff.TrieNodeBytes
targetMetrics.ContractCount += diff.ContractCount
targetMetrics.ContractBytes += diff.ContractBytes
// Fire the metrics and persist only if initialization is done
if initialized {
g.updateMetrics()
g.persistMetrics()
}
}
// calculateUpdateDiff calculates the diff for a state update
func (g *StateSizeGenerator) calculateUpdateDiff(update *stateUpdate) stateSizeMetrics {
var diff stateSizeMetrics
// Calculate account changes
for addr, oldValue := range update.accountsOrigin {
addrHash := crypto.Keccak256Hash(addr.Bytes())
newValue, exists := update.accounts[addrHash]
if !exists {
log.Warn("State update missing account", "address", addr)
continue
}
oldLen, newLen := len(oldValue), len(newValue)
if oldLen > 0 && newLen == 0 {
// Account deletion
diff.AccountCount -= 1
diff.AccountBytes -= accountSnapKeySize + int64(oldLen)
} else if oldLen == 0 && newLen > 0 {
// Account creation
diff.AccountCount += 1
diff.AccountBytes += accountSnapKeySize + int64(newLen)
} else {
// Account update
diff.AccountBytes += int64(newLen - oldLen)
}
}
// Calculate storage changes
for addr, slots := range update.storagesOrigin {
addrHash := crypto.Keccak256Hash(addr.Bytes())
subset, exists := update.storages[addrHash]
if !exists {
log.Warn("State update missing storage", "address", addr)
continue
}
for key, oldValue := range slots {
var (
exists bool
newValue []byte
)
if update.rawStorageKey {
newValue, exists = subset[crypto.Keccak256Hash(key.Bytes())]
} else {
newValue, exists = subset[key]
}
if !exists {
log.Warn("State update missing storage slot", "address", addr, "key", key)
continue
}
oldLen, newLen := len(oldValue), len(newValue)
if oldLen > 0 && newLen == 0 {
// Storage deletion
diff.StorageCount -= 1
diff.StorageBytes -= storageSnapKeySize + int64(oldLen)
} else if oldLen == 0 && newLen > 0 {
// Storage creation
diff.StorageCount += 1
diff.StorageBytes += storageSnapKeySize + int64(newLen)
} else {
// Storage update
diff.StorageBytes += int64(newLen - oldLen)
}
}
}
// Calculate trie node changes
for owner, subset := range update.nodes.Sets {
isAccountTrie := owner == (common.Hash{})
var keyPrefixSize int64
if isAccountTrie {
keyPrefixSize = accountTrieKeyPrefixSize
} else {
keyPrefixSize = storageTrieKeyPrefixSize
}
// Iterate over Origins since every modified node has an origin entry
for path, oldNode := range subset.Origins {
newNode, hasNew := subset.Nodes[path]
keySize := keyPrefixSize + int64(len(path))
if len(oldNode) > 0 && (!hasNew || len(newNode.Blob) == 0) {
// Node deletion
diff.TrieNodeCount -= 1
diff.TrieNodeBytes -= keySize + int64(len(oldNode))
} else if len(oldNode) == 0 && hasNew && len(newNode.Blob) > 0 {
// Node creation
diff.TrieNodeCount += 1
diff.TrieNodeBytes += keySize + int64(len(newNode.Blob))
} else if len(oldNode) > 0 && hasNew && len(newNode.Blob) > 0 {
// Node update
diff.TrieNodeBytes += int64(len(newNode.Blob) - len(oldNode))
}
}
}
// Calculate code changes
for _, code := range update.codes {
diff.ContractCount += 1
diff.ContractBytes += codeKeySize + int64(len(code.blob))
}
return diff
}
// Track is an async method used to send the state update to the generator.
// It ignores empty updates (where no state changes occurred).
// If the channel is full, it drops the update to avoid blocking.
func (g *StateSizeGenerator) Track(update *stateUpdate) {
if update == nil || update.empty() {
return
}
g.updateChan <- update
}
// hasExistingMetrics checks if state size metrics already exist in the database
// and if they are continuous with the current root
func (g *StateSizeGenerator) hasExistingMetrics() bool {
data := rawdb.ReadStateSizeMetrics(g.db)
if data == nil {
return false
}
var existed stateSizeMetrics
if err := json.Unmarshal(data, &existed); err != nil {
log.Warn("Failed to decode existed state size metrics", "err", err)
return false
}
// Check if the existing metrics root matches our current root
if (g.metrics.Root != common.Hash{}) && existed.Root != g.metrics.Root {
log.Info("Existing state size metrics found but root mismatch", "existed", existed.Root, "current", g.metrics.Root)
return false
}
// Root matches - load the existing metrics
log.Info("Loading existing state size metrics", "root", existed.Root)
g.metrics = &existed
return true
}
// initializeMetrics performs the actual metrics initialization using errgroup
func (g *StateSizeGenerator) initializeMetrics() error {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
select {
case <-g.abort:
cancel() // Cancel context when abort is signaled
case <-ctx.Done():
// Context already cancelled
}
}()
// Create errgroup with context
group, ctx := errgroup.WithContext(ctx)
// Metrics will be directly updated by each goroutine
var (
accountSnapCount, accountSnapBytes int64
storageSnapCount, storageSnapBytes int64
accountTrieCount, accountTrieBytes int64
storageTrieCount, storageTrieBytes int64
contractCount, contractBytes int64
)
// Start all table iterations concurrently with direct metric updates
group.Go(func() error {
count, bytes, err := g.iterateTable(ctx, rawdb.SnapshotAccountPrefix, "accountSnap")
if err != nil {
return err
}
accountSnapCount, accountSnapBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := g.iterateTable(ctx, rawdb.SnapshotStoragePrefix, "storageSnap")
if err != nil {
return err
}
storageSnapCount, storageSnapBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := g.iterateTable(ctx, rawdb.TrieNodeAccountPrefix, "accountTrie")
if err != nil {
return err
}
accountTrieCount, accountTrieBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := g.iterateTable(ctx, rawdb.TrieNodeStoragePrefix, "storageTrie")
if err != nil {
return err
}
storageTrieCount, storageTrieBytes = count, bytes
return nil
})
group.Go(func() error {
count, bytes, err := g.iterateTable(ctx, rawdb.CodePrefix, "contract")
if err != nil {
return err
}
contractCount, contractBytes = count, bytes
return nil
})
// Wait for all goroutines to complete
if err := group.Wait(); err != nil {
return err
}
g.metrics.AccountCount = accountSnapCount
g.metrics.AccountBytes = accountSnapBytes
g.metrics.StorageCount = storageSnapCount
g.metrics.StorageBytes = storageSnapBytes
g.metrics.TrieNodeCount = accountTrieCount + storageTrieCount
g.metrics.TrieNodeBytes = accountTrieBytes + storageTrieBytes
g.metrics.ContractCount = contractCount
g.metrics.ContractBytes = contractBytes
g.updateMetrics()
g.persistMetrics()
return nil
}
// iterateTable performs iteration over a specific table and returns the results
func (g *StateSizeGenerator) iterateTable(ctx context.Context, prefix []byte, name string) (int64, int64, error) {
log.Info("Iterating over state size", "table", name)
start := time.Now()
var count, bytes int64
iter := g.db.NewIterator(prefix, nil)
defer iter.Release()
for iter.Next() {
count++
bytes += int64(len(iter.Key()) + len(iter.Value()))
// Check for cancellation periodically for performance
if count%10000 == 0 {
select {
case <-ctx.Done():
log.Info("State size iteration cancelled", "table", name, "count", count)
return 0, 0, ctx.Err()
default:
}
}
}
// Check for iterator errors
if err := iter.Error(); err != nil {
log.Error("Iterator error during state size calculation", "table", name, "err", err)
return 0, 0, err
}
log.Info("Finished iterating over state size", "table", name, "count", count, "bytes", bytes, "elapsed", common.PrettyDuration(time.Since(start)))
return count, bytes, nil
}
func (g *StateSizeGenerator) updateMetrics() {
accountCountGauge.Update(g.metrics.AccountCount)
accountBytesGauge.Update(g.metrics.AccountBytes)
storageCountGauge.Update(g.metrics.StorageCount)
storageBytesGauge.Update(g.metrics.StorageBytes)
trienodeCountGauge.Update(g.metrics.TrieNodeCount)
trienodeBytesGauge.Update(g.metrics.TrieNodeBytes)
contractCountGauge.Update(g.metrics.ContractCount)
contractBytesGauge.Update(g.metrics.ContractBytes)
}
// persistMetrics saves the current metrics to the database
func (g *StateSizeGenerator) persistMetrics() {
// RLP doesn't support int64, so we use JSON for simplicity
data, err := json.Marshal(*g.metrics)
if err != nil {
log.Error("Failed to encode state size metrics", "err", err)
return
}
batch := g.db.NewBatch()
rawdb.WriteStateSizeMetrics(batch, data)
if err := batch.Write(); err != nil {
log.Error("Failed to persist state size metrics", "err", err)
}
}

9
core/state/stateupdate.go
View file

@ -64,8 +64,10 @@ type accountUpdate struct {
// execution. It contains information about mutated contract codes, accounts,
// and storage slots, along with their original values.
type stateUpdate struct {
originRoot common.Hash // hash of the state before applying mutation
root common.Hash // hash of the state after applying mutation
originRoot common.Hash // hash of the state before applying mutation
root common.Hash // hash of the state after applying mutation
blockNumber uint64 // Associated block number
accounts map[common.Hash][]byte // accounts stores mutated accounts in 'slim RLP' encoding
accountsOrigin map[common.Address][]byte // accountsOrigin stores the original values of mutated accounts in 'slim RLP' encoding
@ -95,7 +97,7 @@ func (sc *stateUpdate) empty() bool {
//
// rawStorageKey is a flag indicating whether to use the raw storage slot key or
// the hash of the slot key for constructing state update object.
func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash, deletes map[common.Hash]*accountDelete, updates map[common.Hash]*accountUpdate, nodes *trienode.MergedNodeSet) *stateUpdate {
func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash, blockNumber uint64, deletes map[common.Hash]*accountDelete, updates map[common.Hash]*accountUpdate, nodes *trienode.MergedNodeSet) *stateUpdate {
var (
accounts = make(map[common.Hash][]byte)
accountsOrigin = make(map[common.Address][]byte)
@ -164,6 +166,7 @@ func newStateUpdate(rawStorageKey bool, originRoot common.Hash, root common.Hash
return &stateUpdate{
originRoot: originRoot,
root: root,
blockNumber: blockNumber,
accounts: accounts,
accountsOrigin: accountsOrigin,
storages: storages,

2
core/state/trie_prefetcher_test.go
View file

@ -73,7 +73,7 @@ func TestVerklePrefetcher(t *testing.T) {
state, err := New(types.EmptyRootHash, sdb)
if err != nil {
t.Fatalf("failed to initialize state: %v", err)
t.Fatalf("failed to init state: %v", err)
}
// Create an account and check if the retrieved balance is correct
addr := testrand.Address()

4
triedb/database.go
View file

@ -377,10 +377,10 @@ func (db *Database) Disk() ethdb.Database {
}
// SnapshotCompleted returns the indicator if the snapshot is completed.
func (db *Database) SnapshotCompleted() (common.Hash, bool) {
func (db *Database) SnapshotCompleted() bool {
pdb, ok := db.backend.(*pathdb.Database)
if !ok {
return common.Hash{}, false
return false
}
return pdb.SnapshotCompleted()
}

15
triedb/pathdb/database.go
View file

@ -683,13 +683,12 @@ func (db *Database) StorageIterator(root common.Hash, account common.Hash, seek
}
// SnapshotCompleted returns the snapshot root if the snapshot generation is completed.
func (db *Database) SnapshotCompleted() (common.Hash, bool) {
if db.waitSync {
return common.Hash{}, false
func (db *Database) SnapshotCompleted() bool {
db.lock.RLock()
wait := db.waitSync
db.lock.RUnlock()
if wait {
return false
}
dl := db.tree.bottom()
if dl.genComplete() {
return dl.rootHash(), true
}
return common.Hash{}, false
return db.tree.bottom().genComplete()
}