core, eth, internal: polish

This commit is contained in:
Gary Rong 2025-09-05 14:38:41 +08:00
parent f7332e9d6e
commit a5048f64de
4 changed files with 80 additions and 45 deletions

View file

@ -536,6 +536,7 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine,
stateSizer, err := state.NewSizeTracker(bc.db, bc.triedb) stateSizer, err := state.NewSizeTracker(bc.db, bc.triedb)
if err == nil { if err == nil {
bc.stateSizer = stateSizer bc.stateSizer = stateSizer
log.Info("Initialized state sizer")
} else { } else {
log.Info("Failed to setup size tracker", "err", err) log.Info("Failed to setup size tracker", "err", err)
} }

View file

@ -67,7 +67,7 @@ type SizeStats struct {
} }
func (s SizeStats) String() string { func (s SizeStats) String() string {
return fmt.Sprintf("Accounts: %d (%s), Storages: %d (%s), AccountTrienodes: %d (%s), StorageTrienodes: %d (%s), ContractCodes: %d (%s)", return fmt.Sprintf("Accounts: %d(%s), Storages: %d(%s), AccountTrienodes: %d(%s), StorageTrienodes: %d(%s), Codes: %d(%s)",
s.Accounts, common.StorageSize(s.AccountBytes), s.Accounts, common.StorageSize(s.AccountBytes),
s.Storages, common.StorageSize(s.StorageBytes), s.Storages, common.StorageSize(s.StorageBytes),
s.AccountTrienodes, common.StorageSize(s.AccountTrienodeBytes), s.AccountTrienodes, common.StorageSize(s.AccountTrienodeBytes),
@ -223,6 +223,12 @@ func calSizeStats(update *stateUpdate) (SizeStats, error) {
return stats, nil return stats, nil
} }
type stateSizeQuery struct {
root *common.Hash // nil means latest
err error // non-nil if the state size is not yet initialized
result chan *SizeStats // nil means the state is unknown
}
// SizeTracker handles the state size initialization and tracks of state size metrics. // SizeTracker handles the state size initialization and tracks of state size metrics.
type SizeTracker struct { type SizeTracker struct {
db ethdb.KeyValueStore db ethdb.KeyValueStore
@ -230,9 +236,7 @@ type SizeTracker struct {
abort chan struct{} abort chan struct{}
aborted chan struct{} aborted chan struct{}
updateCh chan *stateUpdate updateCh chan *stateUpdate
queryCh chan *stateSizeQuery
mu sync.RWMutex
latestStats *SizeStats
} }
// NewSizeTracker creates a new state size tracker and starts it automatically // NewSizeTracker creates a new state size tracker and starts it automatically
@ -246,6 +250,7 @@ func NewSizeTracker(db ethdb.KeyValueStore, triedb *triedb.Database) (*SizeTrack
abort: make(chan struct{}), abort: make(chan struct{}),
aborted: make(chan struct{}), aborted: make(chan struct{}),
updateCh: make(chan *stateUpdate), updateCh: make(chan *stateUpdate),
queryCh: make(chan *stateSizeQuery),
} }
go t.run() go t.run()
return t, nil return t, nil
@ -280,6 +285,7 @@ func (h *sizeStatsHeap) Pop() any {
func (t *SizeTracker) run() { func (t *SizeTracker) run() {
defer close(t.aborted) defer close(t.aborted)
var last common.Hash
stats, err := t.init() // launch background thread for state size init stats, err := t.init() // launch background thread for state size init
if err != nil { if err != nil {
return return
@ -301,18 +307,27 @@ func (t *SizeTracker) run() {
} }
stat := base.add(diff) stat := base.add(diff)
stats[u.root] = stat stats[u.root] = stat
log.Debug("Update state size", "number", stat.BlockNumber, "root", stat.StateRoot, "stat", stat) last = u.root
// Update latest stats
t.mu.Lock()
t.latestStats = &stat
t.mu.Unlock()
heap.Push(&h, stats[u.root]) heap.Push(&h, stats[u.root])
for u.blockNumber-h[0].BlockNumber > statEvictThreshold { for u.blockNumber-h[0].BlockNumber > statEvictThreshold {
delete(stats, h[0].StateRoot) delete(stats, h[0].StateRoot)
heap.Pop(&h) heap.Pop(&h)
} }
log.Debug("Update state size", "number", stat.BlockNumber, "root", stat.StateRoot, "stat", stat)
case r := <-t.queryCh:
var root common.Hash
if r.root != nil {
root = *r.root
} else {
root = last
}
if s, ok := stats[root]; ok {
r.result <- &s
} else {
r.result <- nil
}
case <-t.abort: case <-t.abort:
return return
@ -342,6 +357,9 @@ wait:
} }
case <-t.updateCh: case <-t.updateCh:
continue continue
case r := <-t.queryCh:
r.err = errors.New("state size is not initialized yet")
r.result <- nil
case <-t.abort: case <-t.abort:
return nil, errors.New("size tracker closed") return nil, errors.New("size tracker closed")
} }
@ -358,6 +376,11 @@ wait:
case u := <-t.updateCh: case u := <-t.updateCh:
updates[u.root] = u updates[u.root] = u
children[u.originRoot] = append(children[u.originRoot], u.root) children[u.originRoot] = append(children[u.originRoot], u.root)
log.Debug("Received state update", "root", u.root, "blockNumber", u.blockNumber)
case r := <-t.queryCh:
r.err = errors.New("state size is not initialized yet")
r.result <- nil
case <-ticker.C: case <-ticker.C:
// Only check timer if build hasn't started yet // Only check timer if build hasn't started yet
@ -407,11 +430,6 @@ wait:
// Set initial latest stats // Set initial latest stats
stats[result.root] = result.stat stats[result.root] = result.stat
t.mu.Lock()
t.latestStats = &result.stat
t.mu.Unlock()
log.Info("Measured persistent state size", "root", result.root, "number", result.blockNumber, "stat", result.stat, "elapsed", common.PrettyDuration(result.elapsed)) log.Info("Measured persistent state size", "root", result.root, "number", result.blockNumber, "stat", result.stat, "elapsed", common.PrettyDuration(result.elapsed))
return stats, nil return stats, nil
@ -508,20 +526,6 @@ func (t *SizeTracker) build(root common.Hash, blockNumber uint64, done chan buil
} }
} }
// 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. // 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) { func (t *SizeTracker) iterateTable(closed chan struct{}, prefix []byte, name string) (int64, int64, error) {
var ( var (
@ -543,10 +547,10 @@ func (t *SizeTracker) iterateTable(closed chan struct{}, prefix []byte, name str
select { select {
case <-closed: case <-closed:
log.Info("State iteration cancelled", "category", name) log.Debug("State iteration cancelled", "category", name)
return 0, 0, errors.New("size tracker closed") return 0, 0, errors.New("size tracker closed")
default: default:
log.Info("Iterating state", "category", name, "count", count, "size", common.StorageSize(bytes)) log.Debug("Iterating state", "category", name, "count", count, "size", common.StorageSize(bytes))
} }
} }
} }
@ -573,8 +577,7 @@ func (t *SizeTracker) iterateTableParallel(closed chan struct{}, prefix []byte,
mu sync.Mutex mu sync.Mutex
) )
group.SetLimit(workers) group.SetLimit(workers)
log.Debug("Starting parallel state iteration", "category", name, "workers", workers)
log.Info("Starting parallel state iteration", "category", name, "workers", workers)
if len(prefix) > 0 { if len(prefix) > 0 {
if blob, err := t.db.Get(prefix); err == nil && len(blob) > 0 { if blob, err := t.db.Get(prefix); err == nil && len(blob) > 0 {
@ -583,7 +586,6 @@ func (t *SizeTracker) iterateTableParallel(closed chan struct{}, prefix []byte,
totalBytes = int64(len(prefix) + len(blob)) totalBytes = int64(len(prefix) + len(blob))
} }
} }
for i := 0; i < 256; i++ { for i := 0; i < 256; i++ {
h := byte(i) h := byte(i)
group.Go(func() error { group.Go(func() error {
@ -601,14 +603,36 @@ func (t *SizeTracker) iterateTableParallel(closed chan struct{}, prefix []byte,
if err := group.Wait(); err != nil { if err := group.Wait(); err != nil {
return 0, 0, err return 0, 0, err
} }
log.Info("Finished parallel state iteration", "category", name, "count", totalCount, "size", common.StorageSize(totalBytes), "elapsed", common.PrettyDuration(time.Since(start))) log.Debug("Finished parallel state iteration", "category", name, "count", totalCount, "size", common.StorageSize(totalBytes), "elapsed", common.PrettyDuration(time.Since(start)))
return totalCount, totalBytes, nil return totalCount, totalBytes, nil
} }
// GetLatestStats returns the latest state size statistics, or nil if not available // Notify is an async method used to send the state update to the size tracker.
func (t *SizeTracker) GetLatestStats() *SizeStats { // It ignores empty updates (where no state changes occurred).
t.mu.RLock() // If the channel is full, it drops the update to avoid blocking.
defer t.mu.RUnlock() func (t *SizeTracker) Notify(update *stateUpdate) {
if update == nil || update.empty() {
return t.latestStats return
}
select {
case t.updateCh <- update:
case <-t.abort:
return
}
}
// Query returns the state size specified by the root, or nil if not available.
// If the root is nil, query the size of latest chain head;
// If the root is non-nil, query the size of the specified state;
func (t *SizeTracker) Query(root *common.Hash) (*SizeStats, error) {
r := &stateSizeQuery{
root: root,
result: make(chan *SizeStats, 1),
}
select {
case <-t.aborted:
return nil, errors.New("state sizer has been closed")
case t.queryCh <- r:
return <-r.result, r.err
}
} }

View file

@ -446,14 +446,23 @@ func (api *DebugAPI) GetTrieFlushInterval() (string, error) {
// StateSize returns the current state size statistics from the state size tracker. // StateSize returns the current state size statistics from the state size tracker.
// Returns an error if the state size tracker is not initialized or if stats are not ready. // Returns an error if the state size tracker is not initialized or if stats are not ready.
func (api *DebugAPI) StateSize() (interface{}, error) { func (api *DebugAPI) StateSize(root *common.Hash) (interface{}, error) {
sizer := api.eth.blockchain.StateSizer() sizer := api.eth.blockchain.StateSizer()
if sizer == nil { if sizer == nil {
return nil, errors.New("state size tracker is not enabled") return nil, errors.New("state size tracker is not enabled")
} }
stats := sizer.GetLatestStats() stats, err := sizer.Query(root)
if err != nil {
return nil, err
}
if stats == nil { if stats == nil {
return nil, errors.New("state size statistics are not ready yet") var s string
if root == nil {
s = "latest"
} else {
s = root.Hex()
}
return nil, fmt.Errorf("state size %s is not available", s)
} }
return map[string]interface{}{ return map[string]interface{}{
"stateRoot": stats.StateRoot, "stateRoot": stats.StateRoot,

View file

@ -471,7 +471,8 @@ web3._extend({
new web3._extend.Method({ new web3._extend.Method({
name: 'stateSize', name: 'stateSize',
call: 'debug_stateSize', call: 'debug_stateSize',
params: 0, params: 1,
inputFormatter:[null],
}), }),
], ],
properties: [] properties: []