forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core/state: expand prefetcher metrics for reads and writes separately

Browse source
This commit is contained in:
Péter Szilágyi 2024-06-10 13:49:58 +03:00
parent 4ac82c2cf0
commit a82ddd43f6
1 changed files with 109 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

154
core/state/trie_prefetcher.go
View file

@ -47,12 +47,20 @@ type triePrefetcher struct {
noreads bool // Whether to ignore state-read-only prefetch requests noreads bool // Whether to ignore state-read-only prefetch requests
deliveryMissMeter metrics.Meter deliveryMissMeter metrics.Meter
accountLoadMeter metrics.Meter
accountDupMeter metrics.Meter accountLoadReadMeter metrics.Meter
accountWasteMeter metrics.Meter accountLoadWriteMeter metrics.Meter
storageLoadMeter metrics.Meter accountDupReadMeter metrics.Meter
storageDupMeter metrics.Meter accountDupWriteMeter metrics.Meter
storageWasteMeter metrics.Meter accountDupCrossMeter metrics.Meter
accountWasteMeter metrics.Meter
storageLoadReadMeter metrics.Meter
storageLoadWriteMeter metrics.Meter
storageDupReadMeter metrics.Meter
storageDupWriteMeter metrics.Meter
storageDupCrossMeter metrics.Meter
storageWasteMeter metrics.Meter
} }
func newTriePrefetcher(db Database, root common.Hash, namespace string, noreads bool) *triePrefetcher { func newTriePrefetcher(db Database, root common.Hash, namespace string, noreads bool) *triePrefetcher {
@ -65,12 +73,20 @@ func newTriePrefetcher(db Database, root common.Hash, namespace string, noreads
noreads: noreads, noreads: noreads,
deliveryMissMeter: metrics.GetOrRegisterMeter(prefix+"/deliverymiss", nil), deliveryMissMeter: metrics.GetOrRegisterMeter(prefix+"/deliverymiss", nil),
accountLoadMeter: metrics.GetOrRegisterMeter(prefix+"/account/load", nil),
accountDupMeter: metrics.GetOrRegisterMeter(prefix+"/account/dup", nil), accountLoadReadMeter: metrics.GetOrRegisterMeter(prefix+"/account/load/read", nil),
accountWasteMeter: metrics.GetOrRegisterMeter(prefix+"/account/waste", nil), accountLoadWriteMeter: metrics.GetOrRegisterMeter(prefix+"/account/load/write", nil),
storageLoadMeter: metrics.GetOrRegisterMeter(prefix+"/storage/load", nil), accountDupReadMeter: metrics.GetOrRegisterMeter(prefix+"/account/dup/read", nil),
storageDupMeter: metrics.GetOrRegisterMeter(prefix+"/storage/dup", nil), accountDupWriteMeter: metrics.GetOrRegisterMeter(prefix+"/account/dup/write", nil),
storageWasteMeter: metrics.GetOrRegisterMeter(prefix+"/storage/waste", nil), accountDupCrossMeter: metrics.GetOrRegisterMeter(prefix+"/account/dup/cross", nil),
accountWasteMeter: metrics.GetOrRegisterMeter(prefix+"/account/waste", nil),
storageLoadReadMeter: metrics.GetOrRegisterMeter(prefix+"/storage/load/read", nil),
storageLoadWriteMeter: metrics.GetOrRegisterMeter(prefix+"/storage/load/write", nil),
storageDupReadMeter: metrics.GetOrRegisterMeter(prefix+"/storage/dup/read", nil),
storageDupWriteMeter: metrics.GetOrRegisterMeter(prefix+"/storage/dup/write", nil),
storageDupCrossMeter: metrics.GetOrRegisterMeter(prefix+"/storage/dup/cross", nil),
storageWasteMeter: metrics.GetOrRegisterMeter(prefix+"/storage/waste", nil),
} }
} }
@ -100,19 +116,31 @@ func (p *triePrefetcher) report() {
fetcher.wait() // ensure the fetcher's idle before poking in its internals fetcher.wait() // ensure the fetcher's idle before poking in its internals
if fetcher.root == p.root { if fetcher.root == p.root {
p.accountLoadMeter.Mark(int64(len(fetcher.seen))) p.accountLoadReadMeter.Mark(int64(len(fetcher.seenRead)))
p.accountDupMeter.Mark(int64(fetcher.dups)) p.accountLoadWriteMeter.Mark(int64(len(fetcher.seenWrite)))
p.accountDupReadMeter.Mark(int64(fetcher.dupsRead))
p.accountDupWriteMeter.Mark(int64(fetcher.dupsWrite))
p.accountDupCrossMeter.Mark(int64(fetcher.dupsCross))
for _, key := range fetcher.used { for _, key := range fetcher.used {
delete(fetcher.seen, string(key)) delete(fetcher.seenRead, string(key))
delete(fetcher.seenWrite, string(key))
} }
p.accountWasteMeter.Mark(int64(len(fetcher.seen))) p.accountWasteMeter.Mark(int64(len(fetcher.seenRead) + len(fetcher.seenWrite)))
} else { } else {
p.storageLoadMeter.Mark(int64(len(fetcher.seen))) p.storageLoadReadMeter.Mark(int64(len(fetcher.seenRead)))
p.storageDupMeter.Mark(int64(fetcher.dups)) p.storageLoadWriteMeter.Mark(int64(len(fetcher.seenWrite)))
p.storageDupReadMeter.Mark(int64(fetcher.dupsRead))
p.storageDupWriteMeter.Mark(int64(fetcher.dupsWrite))
p.storageDupCrossMeter.Mark(int64(fetcher.dupsCross))
for _, key := range fetcher.used { for _, key := range fetcher.used {
delete(fetcher.seen, string(key)) delete(fetcher.seenRead, string(key))
delete(fetcher.seenWrite, string(key))
} }
p.storageWasteMeter.Mark(int64(len(fetcher.seen))) p.storageWasteMeter.Mark(int64(len(fetcher.seenRead) + len(fetcher.seenWrite)))
} }
} }
} }
@ -145,7 +173,7 @@ func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr comm
fetcher = newSubfetcher(p.db, p.root, owner, root, addr) fetcher = newSubfetcher(p.db, p.root, owner, root, addr)
p.fetchers[id] = fetcher p.fetchers[id] = fetcher
} }
return fetcher.schedule(keys) return fetcher.schedule(keys, read)
} }
// trie returns the trie matching the root hash, blocking until the fetcher of // trie returns the trie matching the root hash, blocking until the fetcher of
@ -192,38 +220,51 @@ type subfetcher struct {
addr common.Address // Address of the account that the trie belongs to addr common.Address // Address of the account that the trie belongs to
trie Trie // Trie being populated with nodes trie Trie // Trie being populated with nodes
tasks [][]byte // Items queued up for retrieval tasks []*subfetcherTask // Items queued up for retrieval
lock sync.Mutex // Lock protecting the task queue lock sync.Mutex // Lock protecting the task queue
wake chan struct{} // Wake channel if a new task is scheduled wake chan struct{} // Wake channel if a new task is scheduled
stop chan struct{} // Channel to interrupt processing stop chan struct{} // Channel to interrupt processing
term chan struct{} // Channel to signal interruption term chan struct{} // Channel to signal interruption
seen map[string]struct{} // Tracks the entries already loaded seenRead map[string]struct{} // Tracks the entries already loaded via read operations
dups int // Number of duplicate preload tasks seenWrite map[string]struct{} // Tracks the entries already loaded via write operations
used [][]byte // Tracks the entries used in the end
dupsRead int // Number of duplicate preload tasks via reads only
dupsWrite int // Number of duplicate preload tasks via writes only
dupsCross int // Number of duplicate preload tasks via read-write-crosses
used [][]byte // Tracks the entries used in the end
}
// subfetcherTask is a trie path to prefetch, tagged with whether it originates
// from a read or a write request.
type subfetcherTask struct {
read bool
key []byte
} }
// newSubfetcher creates a goroutine to prefetch state items belonging to a // newSubfetcher creates a goroutine to prefetch state items belonging to a
// particular root hash. // particular root hash.
func newSubfetcher(db Database, state common.Hash, owner common.Hash, root common.Hash, addr common.Address) *subfetcher { func newSubfetcher(db Database, state common.Hash, owner common.Hash, root common.Hash, addr common.Address) *subfetcher {
sf := &subfetcher{ sf := &subfetcher{
db: db, db: db,
state: state, state: state,
owner: owner, owner: owner,
root: root, root: root,
addr: addr, addr: addr,
wake: make(chan struct{}, 1), wake: make(chan struct{}, 1),
stop: make(chan struct{}), stop: make(chan struct{}),
term: make(chan struct{}), term: make(chan struct{}),
seen: make(map[string]struct{}), seenRead: make(map[string]struct{}),
seenWrite: make(map[string]struct{}),
} }
go sf.loop() go sf.loop()
return sf return sf
} }
// schedule adds a batch of trie keys to the queue to prefetch. // schedule adds a batch of trie keys to the queue to prefetch.
func (sf *subfetcher) schedule(keys [][]byte) error { func (sf *subfetcher) schedule(keys [][]byte, read bool) error {
// Ensure the subfetcher is still alive // Ensure the subfetcher is still alive
select { select {
case <-sf.term: case <-sf.term:
@ -232,7 +273,10 @@ func (sf *subfetcher) schedule(keys [][]byte) error {
} }
// Append the tasks to the current queue // Append the tasks to the current queue
sf.lock.Lock() sf.lock.Lock()
sf.tasks = append(sf.tasks, keys...) for _, key := range keys {
key := key // closure for the append below
sf.tasks = append(sf.tasks, &subfetcherTask{read: read, key: key})
}
sf.lock.Unlock() sf.lock.Unlock()
// Notify the background thread to execute scheduled tasks // Notify the background thread to execute scheduled tasks
@ -309,16 +353,36 @@ func (sf *subfetcher) loop() {
sf.lock.Unlock() sf.lock.Unlock()
for _, task := range tasks { for _, task := range tasks {
if _, ok := sf.seen[string(task)]; ok { key := string(task.key)
sf.dups++ if task.read {
continue if _, ok := sf.seenRead[key]; ok {
} sf.dupsRead++
if len(task) == common.AddressLength { continue
sf.trie.GetAccount(common.BytesToAddress(task)) }
if _, ok := sf.seenWrite[key]; ok {
sf.dupsCross++
continue
}
} else { } else {
sf.trie.GetStorage(sf.addr, task) if _, ok := sf.seenRead[key]; ok {
sf.dupsCross++
continue
}
if _, ok := sf.seenWrite[key]; ok {
sf.dupsWrite++
continue
}
}
if len(task.key) == common.AddressLength {
sf.trie.GetAccount(common.BytesToAddress(task.key))
} else {
sf.trie.GetStorage(sf.addr, task.key)
}
if task.read {
sf.seenRead[key] = struct{}{}
} else {
sf.seenWrite[key] = struct{}{}
} }
sf.seen[string(task)] = struct{}{}
} }
case <-sf.stop: case <-sf.stop: