core/state: don't sort snapshot ordering

This commit is contained in:
Péter Szilágyi 2019-08-08 10:08:20 +03:00
parent 4fc93431c8
commit a86220dfc3
No known key found for this signature in database
GPG key ID: E9AE538CEDF8293D
6 changed files with 108 additions and 84 deletions

View file

@ -51,18 +51,20 @@ var (
headHeaderGauge = metrics.NewRegisteredGauge("chain/head/header", nil) headHeaderGauge = metrics.NewRegisteredGauge("chain/head/header", nil)
headFastBlockGauge = metrics.NewRegisteredGauge("chain/head/receipt", nil) headFastBlockGauge = metrics.NewRegisteredGauge("chain/head/receipt", nil)
accountSnapReadTimer = metrics.NewRegisteredTimer("chain/account/snapreads", nil) accountReadTimer = metrics.NewRegisteredTimer("chain/account/reads", nil)
accountTrieReadTimer = metrics.NewRegisteredTimer("chain/account/triereads", nil)
accountHashTimer = metrics.NewRegisteredTimer("chain/account/hashes", nil) accountHashTimer = metrics.NewRegisteredTimer("chain/account/hashes", nil)
accountUpdateTimer = metrics.NewRegisteredTimer("chain/account/updates", nil) accountUpdateTimer = metrics.NewRegisteredTimer("chain/account/updates", nil)
accountCommitTimer = metrics.NewRegisteredTimer("chain/account/commits", nil) accountCommitTimer = metrics.NewRegisteredTimer("chain/account/commits", nil)
storageSnapReadTimer = metrics.NewRegisteredTimer("chain/storage/snapreads", nil) storageReadTimer = metrics.NewRegisteredTimer("chain/storage/reads", nil)
storageTrieReadTimer = metrics.NewRegisteredTimer("chain/storage/triereads", nil)
storageHashTimer = metrics.NewRegisteredTimer("chain/storage/hashes", nil) storageHashTimer = metrics.NewRegisteredTimer("chain/storage/hashes", nil)
storageUpdateTimer = metrics.NewRegisteredTimer("chain/storage/updates", nil) storageUpdateTimer = metrics.NewRegisteredTimer("chain/storage/updates", nil)
storageCommitTimer = metrics.NewRegisteredTimer("chain/storage/commits", nil) storageCommitTimer = metrics.NewRegisteredTimer("chain/storage/commits", nil)
snapshotAccountReadTimer = metrics.NewRegisteredTimer("chain/snapshot/accountreads", nil)
snapshotStorageReadTimer = metrics.NewRegisteredTimer("chain/snapshot/storagereads", nil)
snapshotCommitTimer = metrics.NewRegisteredTimer("chain/snapshot/commits", nil)
blockInsertTimer = metrics.NewRegisteredTimer("chain/inserts", nil) blockInsertTimer = metrics.NewRegisteredTimer("chain/inserts", nil)
blockValidationTimer = metrics.NewRegisteredTimer("chain/validation", nil) blockValidationTimer = metrics.NewRegisteredTimer("chain/validation", nil)
blockExecutionTimer = metrics.NewRegisteredTimer("chain/execution", nil) blockExecutionTimer = metrics.NewRegisteredTimer("chain/execution", nil)
@ -1630,16 +1632,16 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
return it.index, events, coalescedLogs, err return it.index, events, coalescedLogs, err
} }
// Update the metrics touched during block processing // Update the metrics touched during block processing
accountSnapReadTimer.Update(statedb.AccountSnapReads) // Account reads are complete, we can mark them accountReadTimer.Update(statedb.AccountReads) // Account reads are complete, we can mark them
accountTrieReadTimer.Update(statedb.AccountTrieReads) // Account reads are complete, we can mark them storageReadTimer.Update(statedb.StorageReads) // Storage reads are complete, we can mark them
storageSnapReadTimer.Update(statedb.StorageSnapReads) // Storage reads are complete, we can mark them
storageTrieReadTimer.Update(statedb.StorageTrieReads) // Storage reads are complete, we can mark them
accountUpdateTimer.Update(statedb.AccountUpdates) // Account updates are complete, we can mark them accountUpdateTimer.Update(statedb.AccountUpdates) // Account updates are complete, we can mark them
storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete, we can mark them storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete, we can mark them
snapshotAccountReadTimer.Update(statedb.SnapshotAccountReads) // Account reads are complete, we can mark them
snapshotStorageReadTimer.Update(statedb.SnapshotStorageReads) // Storage reads are complete, we can mark them
triehash := statedb.AccountHashes + statedb.StorageHashes // Save to not double count in validation triehash := statedb.AccountHashes + statedb.StorageHashes // Save to not double count in validation
trieproc := statedb.AccountSnapReads + statedb.AccountTrieReads + statedb.AccountUpdates trieproc := statedb.SnapshotAccountReads + statedb.AccountReads + statedb.AccountUpdates
trieproc += statedb.StorageSnapReads + statedb.StorageTrieReads + statedb.StorageUpdates trieproc += statedb.SnapshotStorageReads + statedb.StorageReads + statedb.StorageUpdates
blockExecutionTimer.Update(time.Since(substart) - trieproc - triehash) blockExecutionTimer.Update(time.Since(substart) - trieproc - triehash)
@ -1670,8 +1672,9 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, []
// Update the metrics touched during block commit // Update the metrics touched during block commit
accountCommitTimer.Update(statedb.AccountCommits) // Account commits are complete, we can mark them accountCommitTimer.Update(statedb.AccountCommits) // Account commits are complete, we can mark them
storageCommitTimer.Update(statedb.StorageCommits) // Storage commits are complete, we can mark them storageCommitTimer.Update(statedb.StorageCommits) // Storage commits are complete, we can mark them
snapshotCommitTimer.Update(statedb.SnapshotCommits) // Snapshot commits are complete, we can mark them
blockWriteTimer.Update(time.Since(substart) - statedb.AccountCommits - statedb.StorageCommits) blockWriteTimer.Update(time.Since(substart) - statedb.AccountCommits - statedb.StorageCommits - statedb.SnapshotCommits)
blockInsertTimer.UpdateSince(start) blockInsertTimer.UpdateSince(start)
switch status { switch status {

View file

@ -43,9 +43,11 @@ type diffLayer struct {
number uint64 // Block number to which this snapshot diff belongs to number uint64 // Block number to which this snapshot diff belongs to
root common.Hash // Root hash to which this snapshot diff belongs to root common.Hash // Root hash to which this snapshot diff belongs to
accountOrder []common.Hash // Sorted accounts for iterated retrieval accountList []common.Hash // List of account for iteration, might not be sorted yet (lazy)
accountSorted bool // Flag whether the account list has alreayd been sorted or not
accountData map[common.Hash][]byte // Keyed accounts for direct retrival (nil means deleted) accountData map[common.Hash][]byte // Keyed accounts for direct retrival (nil means deleted)
storageOrder map[common.Hash][]common.Hash // Sorted storage slots for iterated retrievals. one per account storageList map[common.Hash][]common.Hash // List of storage slots for iterated retrievals, one per account
storageSorted map[common.Hash]bool // Flag whether the storage slot list has alreayd been sorted or not
storageData map[common.Hash]map[common.Hash][]byte // Keyed storage slots for direct retrival. one per account (nil means deleted) storageData map[common.Hash]map[common.Hash][]byte // Keyed storage slots for direct retrival. one per account (nil means deleted)
lock sync.RWMutex lock sync.RWMutex
@ -65,16 +67,21 @@ func newDiffLayer(parent snapshot, root common.Hash, accounts map[common.Hash][]
storageData: storage, storageData: storage,
} }
// Fill the account hashes and sort them for the iterator // Fill the account hashes and sort them for the iterator
dl.accountOrder = make([]common.Hash, 0, len(accounts)) accountList := make([]common.Hash, 0, len(accounts))
for hash, data := range accounts { for hash, data := range accounts {
dl.accountOrder = append(dl.accountOrder, hash) accountList = append(accountList, hash)
dl.memory += uint64(len(data)) dl.memory += uint64(len(data))
} }
sort.Sort(hashes(dl.accountOrder)) sort.Sort(hashes(accountList))
dl.memory += uint64(len(dl.accountOrder) * common.HashLength) dl.accountList = accountList
dl.accountSorted = true
dl.memory += uint64(len(dl.accountList) * common.HashLength)
// Fill the storage hashes and sort them for the iterator // Fill the storage hashes and sort them for the iterator
dl.storageOrder = make(map[common.Hash][]common.Hash, len(storage)) dl.storageList = make(map[common.Hash][]common.Hash, len(storage))
dl.storageSorted = make(map[common.Hash]bool, len(storage))
for accountHash, slots := range storage { for accountHash, slots := range storage {
// If the slots are nil, sanity check that it's a deleted account // If the slots are nil, sanity check that it's a deleted account
if slots == nil { if slots == nil {
@ -83,7 +90,7 @@ func newDiffLayer(parent snapshot, root common.Hash, accounts map[common.Hash][]
panic(fmt.Sprintf("storage in %#x nil, but account conflicts (%#x, exists: %v)", accountHash, account, ok)) panic(fmt.Sprintf("storage in %#x nil, but account conflicts (%#x, exists: %v)", accountHash, account, ok))
} }
// Everything ok, store the deletion mark and continue // Everything ok, store the deletion mark and continue
dl.storageOrder[accountHash] = nil dl.storageList[accountHash] = nil
continue continue
} }
// Storage slots are not nil so entire contract was not deleted, ensure the // Storage slots are not nil so entire contract was not deleted, ensure the
@ -93,16 +100,18 @@ func newDiffLayer(parent snapshot, root common.Hash, accounts map[common.Hash][]
//panic(fmt.Sprintf("storage in %#x exists, but account nil (exists: %v)", accountHash, ok)) //panic(fmt.Sprintf("storage in %#x exists, but account nil (exists: %v)", accountHash, ok))
} }
// Fill the storage hashes for this account and sort them for the iterator // Fill the storage hashes for this account and sort them for the iterator
storageOrder := make([]common.Hash, 0, len(slots)) storageList := make([]common.Hash, 0, len(slots))
for storageHash, data := range slots { for storageHash, data := range slots {
storageOrder = append(storageOrder, storageHash) storageList = append(storageList, storageHash)
dl.memory += uint64(len(data)) dl.memory += uint64(len(data))
} }
sort.Sort(hashes(storageOrder)) sort.Sort(hashes(storageList))
dl.storageOrder[accountHash] = storageOrder dl.storageList[accountHash] = storageList
dl.memory += uint64(len(storageOrder) * common.HashLength) dl.storageSorted[accountHash] = true
dl.memory += uint64(len(storageList) * common.HashLength)
} }
dl.memory += uint64(len(dl.storageOrder) * common.HashLength) dl.memory += uint64(len(dl.storageList) * common.HashLength)
return dl return dl
} }
@ -206,9 +215,6 @@ func (dl *diffLayer) Update(blockRoot common.Hash, accounts map[common.Hash][]by
// the layer limit is reached, memory cap is also enforced (but not before). The // the layer limit is reached, memory cap is also enforced (but not before). The
// block numbers for the disk layer and first diff layer are returned for GC. // block numbers for the disk layer and first diff layer are returned for GC.
func (dl *diffLayer) Cap(layers int, memory uint64) (uint64, uint64) { func (dl *diffLayer) Cap(layers int, memory uint64) (uint64, uint64) {
dl.lock.Lock()
defer dl.lock.Unlock()
// Dive until we run out of layers or reach the persistent database // Dive until we run out of layers or reach the persistent database
if layers > 2 { if layers > 2 {
// If we still have diff layers below, recurse // If we still have diff layers below, recurse
@ -224,6 +230,9 @@ func (dl *diffLayer) Cap(layers int, memory uint64) (uint64, uint64) {
case *diskLayer: case *diskLayer:
return parent.number, dl.number return parent.number, dl.number
case *diffLayer: case *diffLayer:
dl.lock.Lock()
defer dl.lock.Unlock()
dl.parent = parent.flatten() dl.parent = parent.flatten()
if dl.parent.(*diffLayer).memory < memory { if dl.parent.(*diffLayer).memory < memory {
diskNumber, _ := parent.parent.Info() diskNumber, _ := parent.parent.Info()
@ -294,14 +303,15 @@ func (dl *diffLayer) flatten() snapshot {
for hash, data := range dl.accountData { for hash, data := range dl.accountData {
parent.accountData[hash] = data parent.accountData[hash] = data
} }
parent.accountOrder = merge(parent.accountOrder, dl.accountOrder) parent.accountList = append(parent.accountList, dl.accountList...) // TODO(karalabe): dedup!!
parent.accountSorted = false
// Overwrite all the updates storage slots (individually) // Overwrite all the updates storage slots (individually)
for accountHash, storage := range dl.storageData { for accountHash, storage := range dl.storageData {
// If storage didn't exist (or was deleted) in the parent; or if the storage // If storage didn't exist (or was deleted) in the parent; or if the storage
// was freshly deleted in the child, overwrite blindly // was freshly deleted in the child, overwrite blindly
if parent.storageData[accountHash] == nil || storage == nil { if parent.storageData[accountHash] == nil || storage == nil {
parent.storageOrder[accountHash] = dl.storageOrder[accountHash] parent.storageList[accountHash] = dl.storageList[accountHash]
parent.storageData[accountHash] = storage parent.storageData[accountHash] = storage
continue continue
} }
@ -310,8 +320,9 @@ func (dl *diffLayer) flatten() snapshot {
for storageHash, data := range storage { for storageHash, data := range storage {
comboData[storageHash] = data comboData[storageHash] = data
} }
parent.storageOrder[accountHash] = merge(parent.storageOrder[accountHash], dl.storageOrder[accountHash])
parent.storageData[accountHash] = comboData parent.storageData[accountHash] = comboData
parent.storageList[accountHash] = append(parent.storageList[accountHash], dl.storageList[accountHash]...) // TODO(karalabe): dedup!!
parent.storageSorted[accountHash] = false
} }
// Return the combo parent // Return the combo parent
parent.number = dl.number parent.number = dl.number

View file

@ -158,9 +158,9 @@ func (st *SnapshotTree) Cap(blockRoot common.Hash, layers int, memory uint64) er
defer st.lock.Unlock() defer st.lock.Unlock()
diskNumber, diffNumber := snap.Cap(layers, memory) diskNumber, diffNumber := snap.Cap(layers, memory)
for hash, snap := range st.layers { for root, snap := range st.layers {
if number, _ := snap.Info(); number != diskNumber && number < diffNumber { if number, _ := snap.Info(); number != diskNumber && number < diffNumber {
delete(st.layers, hash) delete(st.layers, root)
} }
} }
return nil return nil

View file

@ -193,22 +193,22 @@ func (s *stateObject) GetCommittedState(db Database, key common.Hash) common.Has
enc []byte enc []byte
err error err error
) )
/* if s.db.snap != nil { if s.db.snap != nil {
if metrics.EnabledExpensive { if metrics.EnabledExpensive {
defer func(start time.Time) { s.db.StorageSnapReads += time.Since(start) }(time.Now()) defer func(start time.Time) { s.db.SnapshotStorageReads += time.Since(start) }(time.Now())
} }
enc = s.db.snap.Storage(s.addrHash, crypto.Keccak256Hash(key[:])) enc = s.db.snap.Storage(s.addrHash, crypto.Keccak256Hash(key[:]))
} else {*/ } else {
// Track the amount of time wasted on reading the storage trie // Track the amount of time wasted on reading the storage trie
if metrics.EnabledExpensive { if metrics.EnabledExpensive {
defer func(start time.Time) { s.db.StorageTrieReads += time.Since(start) }(time.Now()) defer func(start time.Time) { s.db.StorageReads += time.Since(start) }(time.Now())
} }
// Otherwise load the value from the database // Otherwise load the value from the database
if enc, err = s.getTrie(db).TryGet(key[:]); err != nil { if enc, err = s.getTrie(db).TryGet(key[:]); err != nil {
s.setError(err) s.setError(err)
return common.Hash{} return common.Hash{}
} }
//} }
if len(enc) > 0 { if len(enc) > 0 {
_, content, _, err := rlp.Split(enc) _, content, _, err := rlp.Split(enc)
if err != nil { if err != nil {

View file

@ -100,16 +100,17 @@ type StateDB struct {
nextRevisionId int nextRevisionId int
// Measurements gathered during execution for debugging purposes // Measurements gathered during execution for debugging purposes
AccountSnapReads time.Duration AccountReads time.Duration
AccountTrieReads time.Duration
AccountHashes time.Duration AccountHashes time.Duration
AccountUpdates time.Duration AccountUpdates time.Duration
AccountCommits time.Duration AccountCommits time.Duration
StorageSnapReads time.Duration StorageReads time.Duration
StorageTrieReads time.Duration
StorageHashes time.Duration StorageHashes time.Duration
StorageUpdates time.Duration StorageUpdates time.Duration
StorageCommits time.Duration StorageCommits time.Duration
SnapshotAccountReads time.Duration
SnapshotStorageReads time.Duration
SnapshotCommits time.Duration
} }
// Create a new state from a given trie. // Create a new state from a given trie.
@ -494,9 +495,9 @@ func (s *StateDB) getStateObject(addr common.Address) (stateObject *stateObject)
} }
// If no live objects are available, attempt to use snapshots // If no live objects are available, attempt to use snapshots
var data Account var data Account
/*if s.snap != nil { if s.snap != nil {
if metrics.EnabledExpensive { if metrics.EnabledExpensive {
defer func(start time.Time) { s.AccountSnapReads += time.Since(start) }(time.Now()) defer func(start time.Time) { s.SnapshotAccountReads += time.Since(start) }(time.Now())
} }
acc := s.snap.Account(crypto.Keccak256Hash(addr[:])) acc := s.snap.Account(crypto.Keccak256Hash(addr[:]))
if acc == nil { if acc == nil {
@ -510,10 +511,10 @@ func (s *StateDB) getStateObject(addr common.Address) (stateObject *stateObject)
if data.Root == (common.Hash{}) { if data.Root == (common.Hash{}) {
data.Root = emptyRoot data.Root = emptyRoot
} }
} else {*/ } else {
// Snapshot unavailable, fall back to the trie // Snapshot unavailable, fall back to the trie
if metrics.EnabledExpensive { if metrics.EnabledExpensive {
defer func(start time.Time) { s.AccountTrieReads += time.Since(start) }(time.Now()) defer func(start time.Time) { s.AccountReads += time.Since(start) }(time.Now())
} }
enc, err := s.trie.TryGet(addr[:]) enc, err := s.trie.TryGet(addr[:])
if len(enc) == 0 { if len(enc) == 0 {
@ -524,7 +525,7 @@ func (s *StateDB) getStateObject(addr common.Address) (stateObject *stateObject)
log.Error("Failed to decode state object", "addr", addr, "err", err) log.Error("Failed to decode state object", "addr", addr, "err", err)
return nil return nil
} }
//} }
// Insert into the live set // Insert into the live set
obj := newObject(s, addr, data) obj := newObject(s, addr, data)
s.setStateObject(obj) s.setStateObject(obj)
@ -768,8 +769,9 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (root common.Hash, err error)
delete(s.stateObjectsDirty, addr) delete(s.stateObjectsDirty, addr)
} }
// Write the account trie changes, measuing the amount of wasted time // Write the account trie changes, measuing the amount of wasted time
var start time.Time
if metrics.EnabledExpensive { if metrics.EnabledExpensive {
defer func(start time.Time) { s.AccountCommits += time.Since(start) }(time.Now()) start = time.Now()
} }
root, err = s.trie.Commit(func(leaf []byte, parent common.Hash) error { root, err = s.trie.Commit(func(leaf []byte, parent common.Hash) error {
var account Account var account Account
@ -785,8 +787,14 @@ func (s *StateDB) Commit(deleteEmptyObjects bool) (root common.Hash, err error)
} }
return nil return nil
}) })
if metrics.EnabledExpensive {
s.AccountCommits += time.Since(start)
}
// If snapshotting is enabled, update the snapshot tree with this new version // If snapshotting is enabled, update the snapshot tree with this new version
if s.snap != nil { if s.snap != nil {
if metrics.EnabledExpensive {
defer func(start time.Time) { s.SnapshotCommits += time.Since(start) }(time.Now())
}
_, parentRoot := s.snap.Info() _, parentRoot := s.snap.Info()
if err := s.snaps.Update(root, parentRoot, s.snapAccounts, s.snapStorage); err != nil { if err := s.snaps.Update(root, parentRoot, s.snapAccounts, s.snapStorage); err != nil {
log.Warn("Failed to update snapshot tree", "from", parentRoot, "to", root, "err", err) log.Warn("Failed to update snapshot tree", "from", parentRoot, "to", root, "err", err)

View file

@ -65,6 +65,8 @@ func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, c
return // Ugh, something went horribly wrong, bail out return // Ugh, something went horribly wrong, bail out
} }
} }
// All transactions processed, finalize the block to force loading written-only trie paths
statedb.Finalise(true) // TODO(karalabe): should we run this on interrupt too?
} }
// precacheTransaction attempts to apply a transaction to the given state database // precacheTransaction attempts to apply a transaction to the given state database