From 462e2bf71b16a20221c540810fe9030f9f40728c Mon Sep 17 00:00:00 2001 From: Fynn Date: Sat, 11 May 2024 14:15:36 +0800 Subject: [PATCH] core,trie,triedb: add getAccount/Storage in pathdb --- core/rawdb/accessors_trie.go | 54 +++++++++ core/rawdb/table.go | 20 +++ core/state/pruner/bloom.go | 5 + core/types/state_account.go | 22 ++++ ethdb/database.go | 3 + ethdb/iterator.go | 3 + ethdb/leveldb/leveldb.go | 10 ++ ethdb/memorydb/memorydb.go | 15 +++ ethdb/pebble/pebble.go | 22 ++++ ethdb/remotedb/remotedb.go | 5 + internal/ethapi/api.go | 5 + trie/node.go | 17 +++ trie/trienode/proof.go | 5 + trie/triestate/state.go | 4 + triedb/hashdb/database.go | 5 + triedb/pathdb/database.go | 7 ++ triedb/pathdb/difflayer.go | 227 +++++++++++++++++++++++++++++++++++ triedb/pathdb/disklayer.go | 124 +++++++++++++++++-- triedb/pathdb/metrics.go | 3 + triedb/pathdb/nodebuffer.go | 82 ++++++++++++- 20 files changed, 620 insertions(+), 18 deletions(-) diff --git a/core/rawdb/accessors_trie.go b/core/rawdb/accessors_trie.go index 44eb715d04..540cdcba5d 100644 --- a/core/rawdb/accessors_trie.go +++ b/core/rawdb/accessors_trie.go @@ -18,6 +18,8 @@ package rawdb import ( "fmt" + "math/big" + "strings" "sync" "github.com/ethereum/go-ethereum/common" @@ -94,6 +96,58 @@ func DeleteAccountTrieNode(db ethdb.KeyValueWriter, path []byte) { } } +func EncodeNibbles(bytes []byte) []byte { + nibbles := make([]byte, len(bytes)*2) + for i, b := range bytes { + nibbles[i*2] = b >> 4 // 取字节高4位 + nibbles[i*2+1] = b & 0x0F // 取字节低4位 + } + return nibbles +} + +func ReadAccountFromTrieDirectly(db ethdb.Database, key []byte) ([]byte, []byte, common.Hash) { + it := db.NewIterator(TrieNodeAccountPrefix, []byte("")) + defer it.Release() + + if it.Seek(accountTrieNodeKey(EncodeNibbles(key))) && it.Error() == nil { + dbKey := common.CopyBytes(it.Key()) + if strings.HasPrefix(string(accountTrieNodeKey(EncodeNibbles(key))), string(dbKey)) { + data := common.CopyBytes(it.Value()) + return data, dbKey[1:], common.Hash{} + } else { + log.Debug("ReadAccountFromTrieDirectly", "dbKey", common.Bytes2Hex(dbKey), "target key", common.Bytes2Hex(accountTrieNodeKey(EncodeNibbles(key)))) + } + } else { + log.Error("ReadAccountFromTrieDirectly", "iterater error", it.Error()) + } + return nil, nil, common.Hash{} +} + +func ReadStorageFromTrieDirectly(db ethdb.Database, accountHash common.Hash, key []byte) ([]byte, []byte, common.Hash) { + it := db.NewIterator(append(TrieNodeStoragePrefix, accountHash.Bytes()...), []byte("")) + defer it.Release() + + if it.Seek(storageTrieNodeKey(accountHash, EncodeNibbles(key))) && it.Error() == nil { + dbKey := common.CopyBytes(it.Key()) + if strings.HasPrefix(string(storageTrieNodeKey(accountHash, EncodeNibbles(key))), string(dbKey)) { + data := common.CopyBytes(it.Value()) + return data, dbKey[1:], common.Hash{} + } + } + return nil, nil, common.Hash{} +} + +func DeleteStorageTrie(db ethdb.KeyValueWriter, accountHash common.Hash) { + nextAccountHash := common.BigToHash(accountHash.Big().Add(accountHash.Big(), big.NewInt(1))) + if err := db.DeleteRange(storageTrieNodeKey(accountHash, nil), storageTrieNodeKey(nextAccountHash, nil)); err != nil { + log.Crit("Failed to delete storage trie", "err", err) + } +} + +func IterateStorageTrieNodes(db ethdb.Iteratee, accountHash common.Hash) ethdb.Iterator { + return db.NewIterator(storageTrieNodeKey(accountHash, nil), nil) +} + // ReadStorageTrieNode retrieves the storage trie node with the specified node path. func ReadStorageTrieNode(db ethdb.KeyValueReader, accountHash common.Hash, path []byte) []byte { data, _ := db.Get(storageTrieNodeKey(accountHash, path)) diff --git a/core/rawdb/table.go b/core/rawdb/table.go index 19e4ed5b5c..2e3f970eed 100644 --- a/core/rawdb/table.go +++ b/core/rawdb/table.go @@ -27,6 +27,11 @@ type table struct { prefix string } +func (t *table) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // NewTable returns a database object that prefixes all keys with a given string. func NewTable(db ethdb.Database, prefix string) ethdb.Database { return &table{ @@ -214,6 +219,11 @@ type tableBatch struct { prefix string } +func (b *tableBatch) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // Put inserts the given value into the batch for later committing. func (b *tableBatch) Put(key, value []byte) error { return b.batch.Put(append([]byte(b.prefix), key...), value) @@ -246,6 +256,11 @@ type tableReplayer struct { prefix string } +func (r *tableReplayer) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // Put implements the interface KeyValueWriter. func (r *tableReplayer) Put(key []byte, value []byte) error { trimmed := key[len(r.prefix):] @@ -270,6 +285,11 @@ type tableIterator struct { prefix string } +func (iter *tableIterator) Seek(key []byte) bool { + // TODO implement me + panic("implement me") +} + // Next moves the iterator to the next key/value pair. It returns whether the // iterator is exhausted. func (iter *tableIterator) Next() bool { diff --git a/core/state/pruner/bloom.go b/core/state/pruner/bloom.go index dad2b5b2a8..1388f551c9 100644 --- a/core/state/pruner/bloom.go +++ b/core/state/pruner/bloom.go @@ -51,6 +51,11 @@ type stateBloom struct { bloom *bloomfilter.Filter } +func (bloom *stateBloom) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // newStateBloomWithSize creates a brand new state bloom for state generation. // The bloom filter will be created by the passing bloom filter size. According // to the https://hur.st/bloomfilter/?n=600000000&p=&m=2048MB&k=4, the parameters diff --git a/core/types/state_account.go b/core/types/state_account.go index 52ef843b35..dc0cb65a13 100644 --- a/core/types/state_account.go +++ b/core/types/state_account.go @@ -18,6 +18,7 @@ package types import ( "bytes" + "fmt" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/rlp" @@ -119,3 +120,24 @@ func FullAccountRLP(data []byte) ([]byte, error) { } return rlp.EncodeToBytes(account) } + +func MustFullAccountRLP(data []byte) []byte { + if data == nil { + return nil + } + val, err := FullAccountRLP(data) + if err != nil { + panic(fmt.Sprintf("must full account rlp faield, err %v", err)) + } + return val +} +func FullToSlimAccountRLP(data []byte) []byte { + if data == nil { + return nil + } + var account StateAccount + if err := rlp.DecodeBytes(data, &account); err != nil { + panic(fmt.Sprintf("full to slim account rlp failed, err %v", err)) + } + return SlimAccountRLP(account) +} diff --git a/ethdb/database.go b/ethdb/database.go index 3ec1f70e3b..5c109e4895 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -35,6 +35,9 @@ type KeyValueWriter interface { // Delete removes the key from the key-value data store. Delete(key []byte) error + + // DeleteRange deletes all of the keys (and values) in the range [start,end) + DeleteRange(start, end []byte) error } // KeyValueStater wraps the Stat method of a backing data store. diff --git a/ethdb/iterator.go b/ethdb/iterator.go index 2b49c93a96..a858662b0e 100644 --- a/ethdb/iterator.go +++ b/ethdb/iterator.go @@ -44,6 +44,9 @@ type Iterator interface { // may change on the next call to Next. Value() []byte + // Seek moves the iterator to the target key/value pair. Only support Lower-bound + Seek(key []byte) bool + // Release releases associated resources. Release should always succeed and can // be called multiple times without causing error. Release() diff --git a/ethdb/leveldb/leveldb.go b/ethdb/leveldb/leveldb.go index e58efbddbe..0958d21dc9 100644 --- a/ethdb/leveldb/leveldb.go +++ b/ethdb/leveldb/leveldb.go @@ -84,6 +84,11 @@ type Database struct { log log.Logger // Contextual logger tracking the database path } +func (db *Database) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // New returns a wrapped LevelDB object. The namespace is the prefix that the // metrics reporting should use for surfacing internal stats. func New(file string, cache int, handles int, namespace string, readonly bool) (*Database, error) { @@ -393,6 +398,11 @@ type batch struct { size int } +func (b *batch) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // Put inserts the given value into the batch for later committing. func (b *batch) Put(key, value []byte) error { b.b.Put(key, value) diff --git a/ethdb/memorydb/memorydb.go b/ethdb/memorydb/memorydb.go index 2a939f9a18..ae0f2c1887 100644 --- a/ethdb/memorydb/memorydb.go +++ b/ethdb/memorydb/memorydb.go @@ -49,6 +49,11 @@ type Database struct { lock sync.RWMutex } +func (db *Database) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // New returns a wrapped map with all the required database interface methods // implemented. func New() *Database { @@ -220,6 +225,11 @@ type batch struct { size int } +func (b *batch) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // Put inserts the given value into the batch for later committing. func (b *batch) Put(key, value []byte) error { b.writes = append(b.writes, keyvalue{string(key), common.CopyBytes(value), false}) @@ -288,6 +298,11 @@ type iterator struct { values [][]byte } +func (it *iterator) Seek(key []byte) bool { + // TODO implement me + panic("implement me") +} + // Next moves the iterator to the next key/value pair. It returns whether the // iterator is exhausted. func (it *iterator) Next() bool { diff --git a/ethdb/pebble/pebble.go b/ethdb/pebble/pebble.go index ee4e5dd75a..b1ce2fc887 100644 --- a/ethdb/pebble/pebble.go +++ b/ethdb/pebble/pebble.go @@ -334,6 +334,17 @@ func (d *Database) Delete(key []byte) error { return d.db.Delete(key, nil) } +// DeleteRange deletes all of the keys (and values) in the range [start,end) +// (inclusive on start, exclusive on end). +func (d *Database) DeleteRange(start, end []byte) error { + d.quitLock.RLock() + defer d.quitLock.RUnlock() + if d.closed { + return pebble.ErrClosed + } + return d.db.DeleteRange(start, end, nil) +} + // NewBatch creates a write-only key-value store that buffers changes to its host // database until a final write is called. func (d *Database) NewBatch() ethdb.Batch { @@ -603,6 +614,13 @@ func (b *batch) Reset() { b.size = 0 } +func (b *batch) DeleteRange(start, end []byte) error { + b.b.DeleteRange(start, end, nil) + b.size += len(start) + b.size += len(end) + return nil +} + // Replay replays the batch contents. func (b *batch) Replay(w ethdb.KeyValueWriter) error { reader := b.b.Reader() @@ -634,6 +652,10 @@ type pebbleIterator struct { released bool } +func (iter *pebbleIterator) Seek(key []byte) bool { + return iter.iter.SeekLT(key) +} + // NewIterator creates a binary-alphabetical iterator over a subset // of database content with a particular key prefix, starting at a particular // initial key (or after, if it does not exist). diff --git a/ethdb/remotedb/remotedb.go b/ethdb/remotedb/remotedb.go index c1c803caf2..e88e5c5357 100644 --- a/ethdb/remotedb/remotedb.go +++ b/ethdb/remotedb/remotedb.go @@ -32,6 +32,11 @@ type Database struct { remote *rpc.Client } +func (db *Database) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + func (db *Database) Has(key []byte) (bool, error) { if _, err := db.Get(key); err != nil { return false, nil diff --git a/internal/ethapi/api.go b/internal/ethapi/api.go index d308cead62..f7e1833953 100644 --- a/internal/ethapi/api.go +++ b/internal/ethapi/api.go @@ -694,6 +694,11 @@ type StorageResult struct { // hex-strings for delivery to rpc-caller. type proofList []string +func (n *proofList) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + func (n *proofList) Put(key []byte, value []byte) error { *n = append(*n, hexutil.Encode(value)) return nil diff --git a/trie/node.go b/trie/node.go index 15bbf62f1c..acb2217ad8 100644 --- a/trie/node.go +++ b/trie/node.go @@ -231,6 +231,23 @@ func decodeRef(buf []byte) (node, []byte, error) { } } +// DecodeLeafNode return the Key and Val part of the shorNode +func DecodeLeafNode(hash, path, value []byte) ([]byte, []byte) { + n := mustDecodeNode(hash, value) + switch sn := n.(type) { + case *shortNode: + if val, ok := sn.Val.(valueNode); ok { + // remove the prefix key of path + key := append(path, sn.Key...) + if hasTerm(key) { + key = key[:len(key)-1] + } + return val, hexToKeybytes(append(path, sn.Key...)) + } + } + return nil, nil +} + // wraps a decoding error with information about the path to the // invalid child node (for debugging encoding issues). type decodeError struct { diff --git a/trie/trienode/proof.go b/trie/trienode/proof.go index 012f0087dd..03b72c67e6 100644 --- a/trie/trienode/proof.go +++ b/trie/trienode/proof.go @@ -36,6 +36,11 @@ type ProofSet struct { lock sync.RWMutex } +func (db *ProofSet) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // NewProofSet creates an empty node set func NewProofSet() *ProofSet { return &ProofSet{ diff --git a/trie/triestate/state.go b/trie/triestate/state.go index 9db9211e8c..4fc7d444d0 100644 --- a/trie/triestate/state.go +++ b/trie/triestate/state.go @@ -58,6 +58,10 @@ type TrieLoader interface { // The value refers to the original content of state before the transition // is made. Nil means that the state was not present previously. type Set struct { + LatestAccounts map[common.Hash][]byte + LatestStorages map[common.Hash]map[common.Hash][]byte + DestructSet map[common.Hash]struct{} + Accounts map[common.Address][]byte // Mutated account set, nil means the account was not present Storages map[common.Address]map[common.Hash][]byte // Mutated storage set, nil means the slot was not present size common.StorageSize // Approximate size of set diff --git a/triedb/hashdb/database.go b/triedb/hashdb/database.go index ebb5d72057..367bd13741 100644 --- a/triedb/hashdb/database.go +++ b/triedb/hashdb/database.go @@ -493,6 +493,11 @@ type cleaner struct { db *Database } +func (c *cleaner) DeleteRange(start, end []byte) error { + // TODO implement me + panic("implement me") +} + // Put reacts to database writes and implements dirty data uncaching. This is the // post-processing step of a commit operation where the already persisted trie is // removed from the dirty cache and moved into the clean cache. The reason behind diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index 05a28aa1ef..b3f3302bd2 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -67,6 +67,13 @@ type layer interface { // Note, no error will be returned if the requested node is not found in database. node(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) + // Account directly retrieves the account data associated with a particular hash + Account(hash common.Hash) ([]byte, error) + + // Storage directly retrieves the storage data associated with a particular hash, + // within a particular account. + Storage(accountHash, storageHash common.Hash) ([]byte, error) + // rootHash returns the root hash for which this layer was made. rootHash() common.Hash diff --git a/triedb/pathdb/difflayer.go b/triedb/pathdb/difflayer.go index 6b87883482..35cb8b26a0 100644 --- a/triedb/pathdb/difflayer.go +++ b/triedb/pathdb/difflayer.go @@ -17,15 +17,95 @@ package pathdb import ( + "encoding/binary" "fmt" + "math" + "math/rand" "sync" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/trie/trienode" "github.com/ethereum/go-ethereum/trie/triestate" + bloomfilter "github.com/holiman/bloomfilter/v2" ) +var ( + // aggregatorMemoryLimit is the maximum size of the bottom-most diff layer + // that aggregates the writes from above until it's flushed into the disk + // layer. + // + // Note, bumping this up might drastically increase the size of the bloom + // filters that's stored in every diff layer. Don't do that without fully + // understanding all the implications. + aggregatorMemoryLimit = uint64(4 * 1024 * 1024) + + // aggregatorItemLimit is an approximate number of items that will end up + // in the aggregator layer before it's flushed out to disk. A plain account + // weighs around 14B (+hash), a storage slot 32B (+hash), a deleted slot + // 0B (+hash). Slots are mostly set/unset in lockstep, so that average at + // 16B (+hash). All in all, the average entry seems to be 15+32=47B. Use a + // smaller number to be on the safe side. + aggregatorItemLimit = aggregatorMemoryLimit / 42 + // bloomTargetError is the target false positive rate when the aggregator + // layer is at its fullest. The actual value will probably move around up + // and down from this number, it's mostly a ballpark figure. + // + // Note, dropping this down might drastically increase the size of the bloom + // filters that's stored in every diff layer. Don't do that without fully + // understanding all the implications. + bloomTargetError = 0.02 + + // bloomSize is the ideal bloom filter size given the maximum number of items + // it's expected to hold and the target false positive error rate. + bloomSize = math.Ceil(float64(aggregatorItemLimit) * math.Log(bloomTargetError) / math.Log(1/math.Pow(2, math.Log(2)))) + + // bloomFuncs is the ideal number of bits a single entry should set in the + // bloom filter to keep its size to a minimum (given it's size and maximum + // entry count). + bloomFuncs = math.Round((bloomSize / float64(aggregatorItemLimit)) * math.Log(2)) + + // the bloom offsets are runtime constants which determines which part of the + // account/storage hash the hasher functions looks at, to determine the + // bloom key for an account/slot. This is randomized at init(), so that the + // global population of nodes do not all display the exact same behaviour with + // regards to bloom content + bloomDestructHasherOffset = 0 + bloomAccountHasherOffset = 0 + bloomStorageHasherOffset = 0 +) + +func init() { + // Init the bloom offsets in the range [0:24] (requires 8 bytes) + bloomDestructHasherOffset = rand.Intn(25) + bloomAccountHasherOffset = rand.Intn(25) + bloomStorageHasherOffset = rand.Intn(25) + + // The destruct and account blooms must be different, as the storage slots + // will check for destruction too for every bloom miss. It should not collide + // with modified accounts. + for bloomAccountHasherOffset == bloomDestructHasherOffset { + bloomAccountHasherOffset = rand.Intn(25) + } +} + +// destructBloomHash is used to convert a destruct event into a 64 bit mini hash. +func destructBloomHash(h common.Hash) uint64 { + return binary.BigEndian.Uint64(h[bloomDestructHasherOffset : bloomDestructHasherOffset+8]) +} + +// accountBloomHash is used to convert an account hash into a 64 bit mini hash. +func accountBloomHash(h common.Hash) uint64 { + return binary.BigEndian.Uint64(h[bloomAccountHasherOffset : bloomAccountHasherOffset+8]) +} + +// storageBloomHash is used to convert an account hash and a storage hash into a 64 bit mini hash. +func storageBloomHash(h0, h1 common.Hash) uint64 { + return binary.BigEndian.Uint64(h0[bloomStorageHasherOffset:bloomStorageHasherOffset+8]) ^ + binary.BigEndian.Uint64(h1[bloomStorageHasherOffset:bloomStorageHasherOffset+8]) +} + // diffLayer represents a collection of modifications made to the in-memory tries // along with associated state changes after running a block on top. // @@ -42,6 +122,10 @@ type diffLayer struct { parent layer // Parent layer modified by this one, never nil, **can be changed** lock sync.RWMutex // Lock used to protect parent + + diffed *bloomfilter.Filter // Bloom filter tracking all the diffed items up to the disk layer + selfDiffed *bloomfilter.Filter // Bloom filter tracking all the diffed items of the diff layer + origin *diskLayer // Base disk layer to directly use on bloom misses } // newDiffLayer creates a new diff layer on top of an existing layer. @@ -58,6 +142,15 @@ func newDiffLayer(parent layer, root common.Hash, id uint64, block uint64, nodes states: states, parent: parent, } + switch l := parent.(type) { + case *diskLayer: + dl.rebloom(l) + case *diffLayer: + dl.rebloom(l.origin) + default: + panic("unknown parent type") + } + for _, subset := range nodes { for path, n := range subset { dl.memory += uint64(n.Size() + len(path)) @@ -75,6 +168,61 @@ func newDiffLayer(parent layer, root common.Hash, id uint64, block uint64, nodes return dl } +// rebloom discards the layer's current bloom and rebuilds it from scratch based +// on the parent's and the local diffs. +func (dl *diffLayer) rebloom(origin *diskLayer) { + dl.lock.Lock() + defer dl.lock.Unlock() + + defer func(start time.Time) { + bloomIndexTimer.Update(time.Since(start)) + }(time.Now()) + + // Inject the new origin that triggered the rebloom + dl.origin = origin + + // Retrieve the parent bloom or create a fresh empty one + if parent, ok := dl.parent.(*diffLayer); ok { + parent.lock.RLock() + dl.diffed, _ = parent.diffed.Copy() + parent.lock.RUnlock() + } else { + if dl.selfDiffed == nil { + dl.diffed, _ = bloomfilter.New(uint64(bloomSize), uint64(bloomFuncs)) + } else { + dl.diffed, _ = dl.selfDiffed.NewCompatible() + } + } + + if dl.selfDiffed == nil { + dl.selfDiffed, _ = dl.diffed.NewCompatible() + // Iterate over all the accounts and storage slots and index them + for h := range dl.states.DestructSet { + dl.selfDiffed.AddHash(destructBloomHash(h)) + } + for h := range dl.states.LatestAccounts { + dl.selfDiffed.AddHash(accountBloomHash(h)) + } + for accountHash, slots := range dl.states.LatestStorages { + for storageHash := range slots { + dl.selfDiffed.AddHash(storageBloomHash(accountHash, storageHash)) + } + } + } + err := dl.diffed.UnionInPlace(dl.selfDiffed) + if err != nil { + log.Error("diff layer bloom filter failed to union in place", "id", dl.id, "err", err) + } + + // Calculate the current false positive rate and update the error rate meter. + // This is a bit cheating because subsequent layers will overwrite it, but it + // should be fine, we're only interested in ballpark figures. + k := float64(dl.diffed.K()) + n := float64(dl.diffed.N()) + m := float64(dl.diffed.M()) + bloomErrorGauge.Update(math.Pow(1.0-math.Exp((-k)*(n+0.5)/(m-1)), k)) +} + // rootHash implements the layer interface, returning the root hash of // corresponding state. func (dl *diffLayer) rootHash() common.Hash { @@ -95,6 +243,85 @@ func (dl *diffLayer) parentLayer() layer { return dl.parent } +func (dl *diffLayer) Account(hash common.Hash) ([]byte, error) { + dl.lock.RLock() + defer dl.lock.RUnlock() + // Check the bloom filter first whether there's even a point in reaching into + // all the maps in all the layers below + hit := dl.diffed.ContainsHash(accountBloomHash(hash)) + if !hit { + hit = dl.diffed.ContainsHash(destructBloomHash(hash)) + } + var origin *diskLayer + if !hit { + origin = dl.origin // extract origin while holding the lock + } + // If the bloom filter misses, don't even bother with traversing the memory + // diff layers, reach straight into the bottom persistent disk layer + if origin != nil { + return origin.Account(hash) + } + + return dl.account(hash) +} + +func (dl *diffLayer) account(hash common.Hash) ([]byte, error) { + dl.lock.RLock() + defer dl.lock.RUnlock() + + if data, ok := dl.states.LatestAccounts[hash]; ok { + return data, nil + } + + if _, ok := dl.states.DestructSet[hash]; ok { + return nil, nil + } + + if diff, ok := dl.parent.(*diffLayer); ok { + return diff.account(hash) + } + return dl.parent.Account(hash) +} + +func (dl *diffLayer) Storage(accountHash, storageHash common.Hash) ([]byte, error) { + dl.lock.RLock() + defer dl.lock.RUnlock() + + hit := dl.diffed.ContainsHash(storageBloomHash(accountHash, storageHash)) + if !hit { + hit = dl.diffed.ContainsHash(destructBloomHash(accountHash)) + } + var origin *diskLayer + if !hit { + origin = dl.origin // extract origin while holding the lock + } + + if origin != nil { + return origin.Storage(accountHash, storageHash) + } + return dl.storage(accountHash, storageHash) +} + +func (dl *diffLayer) storage(accountHash, storageHash common.Hash) ([]byte, error) { + dl.lock.RLock() + defer dl.lock.RUnlock() + + if storage, ok := dl.states.LatestStorages[accountHash]; ok { + if data, ok := storage[storageHash]; ok { + return data, nil + } + } + + if _, ok := dl.states.DestructSet[accountHash]; ok { + return nil, nil + } + // Storage slot unknown to this diff, resolve from parent + if diff, ok := dl.parent.(*diffLayer); ok { + return diff.storage(accountHash, storageHash) + } + return dl.parent.Storage(accountHash, storageHash) +} + // node implements the layer interface, retrieving the trie node blob with the // provided node information. No error will be returned if the node is not found. func (dl *diffLayer) node(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) { diff --git a/triedb/pathdb/disklayer.go b/triedb/pathdb/disklayer.go index 964ad2ef77..45b14d79b1 100644 --- a/triedb/pathdb/disklayer.go +++ b/triedb/pathdb/disklayer.go @@ -17,36 +17,60 @@ package pathdb import ( + "bytes" + "encoding/hex" "fmt" "sync" "github.com/VictoriaMetrics/fastcache" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie/trienode" "github.com/ethereum/go-ethereum/trie/triestate" ) // diskLayer is a low level persistent layer built on top of a key-value store. type diskLayer struct { - root common.Hash // Immutable, root hash to which this layer was made for - id uint64 // Immutable, corresponding state id - db *Database // Path-based trie database - cleans *fastcache.Cache // GC friendly memory cache of clean node RLPs - buffer *nodebuffer // Node buffer to aggregate writes - stale bool // Signals that the layer became stale (state progressed) - lock sync.RWMutex // Lock used to protect stale flag + root common.Hash // Immutable, root hash to which this layer was made for + id uint64 // Immutable, corresponding state id + db *Database // Path-based trie database + cleans *cleanCache // GC friendly memory cache of clean node RLPs + buffer *nodebuffer // Node buffer to aggregate writes + stale bool // Signals that the layer became stale (state progressed) + lock sync.RWMutex // Lock used to protect stale flag +} + +type cleanCache struct { + nodes *fastcache.Cache + plainStates *fastcache.Cache +} + +func (cc *cleanCache) reset() { + if cc.nodes != nil { + cc.nodes.Reset() + } + if cc.plainStates != nil { + cc.plainStates.Reset() + } } // newDiskLayer creates a new disk layer based on the passing arguments. -func newDiskLayer(root common.Hash, id uint64, db *Database, cleans *fastcache.Cache, buffer *nodebuffer) *diskLayer { +func newDiskLayer(root common.Hash, id uint64, db *Database, cleans *cleanCache, buffer *nodebuffer) *diskLayer { // Initialize a clean cache if the memory allowance is not zero // or reuse the provided cache if it is not nil (inherited from // the original disk layer). if cleans == nil && db.config.CleanCacheSize != 0 { - cleans = fastcache.New(db.config.CleanCacheSize) + cleans = &cleanCache{} + nodeCacheSize := db.config.CleanCacheSize * 43 / 100 + plainStatesCacheSize := db.config.CleanCacheSize * 57 / 100 + cleans.nodes = fastcache.New(nodeCacheSize) + cleans.plainStates = fastcache.New(plainStatesCacheSize) + log.Info("Allocate clean cache in disklayer", "nodes", common.StorageSize(nodeCacheSize), + "plainStates", common.StorageSize(plainStatesCacheSize)) } return &diskLayer{ root: root, @@ -93,6 +117,82 @@ func (dl *diskLayer) markStale() { dl.stale = true } +func (dl *diskLayer) Account(hash common.Hash) ([]byte, error) { + // Hold the lock, ensure the parent won't be changed during the + // state accessing. + dl.lock.RLock() + defer dl.lock.RUnlock() + if dl.stale { + return nil, errSnapshotStale + } + + if data, exist := dl.buffer.account(hash); exist { + return data, nil + } + if data, ok := dl.cleans.plainStates.HasGet(nil, hash.Bytes()); ok { + return types.MustFullAccountRLP(data), nil + } + + blob := dl.readAccountTrie(hash) + dl.cleans.plainStates.Set(hash.Bytes(), types.FullToSlimAccountRLP(blob)) + return blob, nil +} + +// readAccountTrie return value of the account leaf node directly from the db +func (dl *diskLayer) readAccountTrie(hash common.Hash) []byte { + nBlob, path, nHash := rawdb.ReadAccountFromTrieDirectly(dl.db.diskdb, hash.Bytes()) + if nBlob == nil { + return nil + } + dl.cleans.nodes.Set(cacheKey(common.Hash{}, path[:]), nBlob) + val, key := trie.DecodeLeafNode(nHash.Bytes(), path, nBlob) + if bytes.Compare(key, hash.Bytes()) == 0 { + return val + } else { + log.Debug("account short node info ", "account hash", hash.String(), "gotten key", hex.EncodeToString(key), "path", common.Bytes2Hex(path)) + } + return nil +} + +func (dl *diskLayer) Storage(accountHash, storageHash common.Hash) ([]byte, error) { + // Hold the lock, ensure the parent won't be changed during the + // state accessing. + dl.lock.RLock() + defer dl.lock.RUnlock() + + if dl.stale { + return nil, errSnapshotStale + } + + if data, exist := dl.buffer.storage(accountHash, storageHash); exist { + return data, nil + } + + if data, ok := dl.cleans.plainStates.HasGet(nil, append(accountHash.Bytes(), storageHash.Bytes()...)); ok { + return data, nil + } + + blob := dl.readStorageTrie(accountHash, storageHash) + dl.cleans.plainStates.Set(append(accountHash.Bytes(), storageHash.Bytes()...), blob) + return blob, nil +} + +// readStorageTrie return value of the storage leaf node directly from the db +func (dl *diskLayer) readStorageTrie(accountHash, storageHash common.Hash) []byte { + key := storageHash.Bytes() + nBlob, path, nHash := rawdb.ReadStorageFromTrieDirectly(dl.db.diskdb, accountHash, key) + if nBlob == nil { + return nil + } + dl.cleans.nodes.Set(cacheKey(accountHash, path[common.HashLength:]), nBlob) + + val, key := trie.DecodeLeafNode(nHash.Bytes(), path[common.HashLength:], nBlob) + if bytes.Compare(storageHash.Bytes(), key) == 0 { + return val + } + return nil +} + // node implements the layer interface, retrieving the trie node with the // provided node info. No error will be returned if the node is not found. func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, common.Hash, *nodeLoc, error) { @@ -121,7 +221,7 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co key := cacheKey(owner, path) if dl.cleans != nil { - if blob := dl.cleans.Get(nil, key); len(blob) > 0 { + if blob := dl.cleans.nodes.Get(nil, key); len(blob) > 0 { cleanHitMeter.Mark(1) cleanReadMeter.Mark(int64(len(blob))) return blob, h.hash(blob), &nodeLoc{loc: locCleanCache, depth: depth}, nil @@ -136,7 +236,7 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co blob = rawdb.ReadStorageTrieNode(dl.db.diskdb, owner, path) } if dl.cleans != nil && len(blob) > 0 { - dl.cleans.Set(key, blob) + dl.cleans.nodes.Set(key, blob) cleanWriteMeter.Mark(int64(len(blob))) } @@ -292,7 +392,7 @@ func (dl *diskLayer) resetCache() { return } if dl.cleans != nil { - dl.cleans.Reset() + dl.cleans.reset() } } diff --git a/triedb/pathdb/metrics.go b/triedb/pathdb/metrics.go index a250f703cb..fcec2b850b 100644 --- a/triedb/pathdb/metrics.go +++ b/triedb/pathdb/metrics.go @@ -48,4 +48,7 @@ var ( historyBuildTimeMeter = metrics.NewRegisteredTimer("pathdb/history/time", nil) historyDataBytesMeter = metrics.NewRegisteredMeter("pathdb/history/bytes/data", nil) historyIndexBytesMeter = metrics.NewRegisteredMeter("pathdb/history/bytes/index", nil) + + bloomIndexTimer = metrics.NewRegisteredResettingTimer("pathdb/bloom/index", nil) + bloomErrorGauge = metrics.NewRegisteredGaugeFloat64("pathdb/bloom/error", nil) ) diff --git a/triedb/pathdb/nodebuffer.go b/triedb/pathdb/nodebuffer.go index ff09484100..873ffacf82 100644 --- a/triedb/pathdb/nodebuffer.go +++ b/triedb/pathdb/nodebuffer.go @@ -19,14 +19,16 @@ package pathdb import ( "bytes" "fmt" + "sync" "time" - "github.com/VictoriaMetrics/fastcache" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie/trienode" ) @@ -38,6 +40,11 @@ type nodebuffer struct { size uint64 // The size of aggregated writes limit uint64 // The maximum memory allowance in bytes nodes map[common.Hash]map[string]*trienode.Node // The dirty node set, mapped by owner and path + + // latest account and storage + LatestAccounts map[common.Hash][]byte + LatestStorages map[common.Hash]map[common.Hash][]byte + DestructSet map[common.Hash]struct{} } // newNodeBuffer initializes the node buffer with the provided nodes. @@ -59,6 +66,30 @@ func newNodeBuffer(limit int, nodes map[common.Hash]map[string]*trienode.Node, l } } +func (b *nodebuffer) account(hash common.Hash) ([]byte, bool) { + if data, ok := b.LatestAccounts[hash]; ok { + return data, true + } + + if _, ok := b.DestructSet[hash]; ok { + return nil, true + } + return nil, false +} + +func (b *nodebuffer) storage(accountHash, storageHash common.Hash) ([]byte, bool) { + if storage, ok := b.LatestStorages[accountHash]; ok { + if data, ok := storage[storageHash]; ok { + return data, true + } + } + + if _, ok := b.DestructSet[accountHash]; ok { + return nil, true + } + return nil, false +} + // node retrieves the trie node with given node info. func (b *nodebuffer) node(owner common.Hash, path []byte) (*trienode.Node, bool) { subset, ok := b.nodes[owner] @@ -195,7 +226,7 @@ func (b *nodebuffer) empty() bool { // setSize sets the buffer size to the provided number, and invokes a flush // operation if the current memory usage exceeds the new limit. -func (b *nodebuffer) setSize(size int, db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64) error { +func (b *nodebuffer) setSize(size int, db ethdb.KeyValueStore, clean *cleanCache, id uint64) error { b.limit = uint64(size) return b.flush(db, clean, id, false) } @@ -215,7 +246,7 @@ func (b *nodebuffer) allocBatch(db ethdb.KeyValueStore) ethdb.Batch { // flush persists the in-memory dirty trie node into the disk if the configured // memory threshold is reached. Note, all data must be written atomically. -func (b *nodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, id uint64, force bool) error { +func (b *nodebuffer) flush(db ethdb.KeyValueStore, clean *cleanCache, id uint64, force bool) error { if b.size <= b.limit && !force { return nil } @@ -228,6 +259,45 @@ func (b *nodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, id ui start = time.Now() batch = b.allocBatch(db) ) + // delete all kv for destructSet first to keep latest for disk nodes + for h, _ := range b.DestructSet { + rawdb.DeleteStorageTrie(batch, h) + clean.plainStates.Set(h.Bytes(), nil) + } + var wg sync.WaitGroup + if len(b.DestructSet) != 0 { + wg.Add(1) + go func() { + st := time.Now() + nums := 0 + for h := range b.DestructSet { + // delete from the clean cache + it := rawdb.IterateStorageTrieNodes(db, h) + for it.Next() { + if it.Value() != nil { + h := newHasher() + _, key := trie.DecodeLeafNode(h.hash(it.Value()).Bytes(), it.Key()[1+common.HashLength:], it.Value()) + if key != nil { + nums++ + clean.plainStates.Del(key) + } + h.release() + } + } + it.Release() + } + log.Info("handle deletion of plain storage", "elapsed", time.Since(st).String(), "deleted nums", nums) + for h, acc := range b.LatestAccounts { + clean.plainStates.Set(h.Bytes(), types.FullToSlimAccountRLP(acc)) + } + for h, storages := range b.LatestStorages { + for k, v := range storages { + clean.plainStates.Set(append(h.Bytes(), k.Bytes()...), v) + } + } + wg.Done() + }() + } nodes := writeNodes(batch, b.nodes, clean) rawdb.WritePersistentStateID(batch, id) @@ -247,7 +317,7 @@ func (b *nodebuffer) flush(db ethdb.KeyValueStore, clean *fastcache.Cache, id ui // writeNodes writes the trie nodes into the provided database batch. // Note this function will also inject all the newly written nodes // into clean cache. -func writeNodes(batch ethdb.Batch, nodes map[common.Hash]map[string]*trienode.Node, clean *fastcache.Cache) (total int) { +func writeNodes(batch ethdb.Batch, nodes map[common.Hash]map[string]*trienode.Node, clean *cleanCache) (total int) { for owner, subset := range nodes { for path, n := range subset { if n.IsDeleted() { @@ -257,7 +327,7 @@ func writeNodes(batch ethdb.Batch, nodes map[common.Hash]map[string]*trienode.No rawdb.DeleteStorageTrieNode(batch, owner, []byte(path)) } if clean != nil { - clean.Del(cacheKey(owner, []byte(path))) + clean.nodes.Del(cacheKey(owner, []byte(path))) } } else { if owner == (common.Hash{}) { @@ -266,7 +336,7 @@ func writeNodes(batch ethdb.Batch, nodes map[common.Hash]map[string]*trienode.No rawdb.WriteStorageTrieNode(batch, owner, []byte(path), n.Blob) } if clean != nil { - clean.Set(cacheKey(owner, []byte(path)), n.Blob) + clean.nodes.Set(cacheKey(owner, []byte(path)), n.Blob) } } }