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

core/state, trie: Refactored direct cache backend to be more coherent.

Browse source
This commit is contained in:
Nick Johnson 2016-10-27 14:10:11 +01:00
parent 21ce20c359
commit 3262254496
7 changed files with 145 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
core/state/iterator_test.go
View file

@ -47,7 +47,7 @@ func TestNodeIteratorCoverage(t *testing.T) {
}
}
for _, key := range db.(*ethdb.MemDatabase).Keys() {
if bytes.HasPrefix(key, []byte("secure-key-")) || bytes.HasPrefix(key, CachePrefix) {
if bytes.HasPrefix(key, []byte("secure-key-")) || bytes.HasPrefix(key, DirectCachePrefix) {
continue
}
if _, ok := hashes[common.BytesToHash(key)]; !ok {

14
core/state/statedb.go
View file

@ -41,7 +41,8 @@ var StartingNonce uint64
// Trie cache generation limit after which to evic trie nodes from memory.
var MaxTrieCacheGen = uint16(120)
var CachePrefix = []byte("accounthashcache:")
var DirectCachePrefix = []byte("accounthashcache:")
var DirectStateCacheCompleteKey = []byte("directstatecache:complete")
const (
// Number of past tries to keep. This value is chosen such that
@ -66,6 +67,7 @@ type StateDB struct {
db ethdb.Database
trie *trie.Trie
storage *trie.SecureTrie
cacheComplete bool // True if we know that the directcache is complete
pastTries []*trie.Trie
codeSizeCache *lru.Cache
@ -188,7 +190,15 @@ func (self *StateDB) SetBlockContext(blockHash common.Hash, blockNum uint64, val
if validator == nil {
validator = &trie.NullCacheValidator{}
}
storage := trie.NewDirectCache(self.trie, self.db, CachePrefix, blockNum, blockHash, validator, true)
// Check if cache population has completed
if !self.cacheComplete {
if value, err := self.db.Get(DirectStateCacheCompleteKey); err == nil && value != nil {
self.cacheComplete = true
}
}
storage := trie.NewDirectCache(self.trie, self.db, DirectCachePrefix, blockNum, blockHash, validator, true)
self.storage = trie.NewSecure(storage, self.db)
}

9
core/state/sync.go
View file

@ -47,18 +47,23 @@ func NewStateSync(number uint64, hash common.Hash, root common.Hash, database et
return err
}
// Populate the direct account caches in the database
if err := trie.DirectCacheTransaction(func() error {
for _, key := range keys {
if err := trie.WriteDirectCache(CachePrefix, key, leaf, number, hash, database); err != nil {
if err := trie.WriteDirectCache(DirectCachePrefix, key, leaf, number, hash, database); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
// Schedule downloading all the dependencies of the account
syncer.AddSubTrie(obj.Root, 64, parent, nil, nil)
syncer.AddRawEntry(common.BytesToHash(obj.CodeHash), 64, parent)
return nil
}
syncer = trie.NewTrieSync(root, database, callback, CachePrefix)
syncer = trie.NewTrieSync(root, database, callback, DirectCachePrefix)
return (*StateSync)(syncer)
}

4
eth/backend.go
View file

@ -235,6 +235,10 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
gpo := gasprice.NewGasPriceOracle(eth.blockchain, chainDb, eth.eventMux, gpoParams)
eth.apiBackend = &EthApiBackend{eth, gpo}
if err := populateDirectCache(chainDb, eth.blockchain); err != nil {
return nil, err
}
return eth, nil
}

7
eth/db_upgrade.go
View file

@ -354,3 +354,10 @@ func addMipmapBloomBins(db ethdb.Database) (err error) {
glog.V(logger.Info).Infoln("upgrade completed in", time.Since(tstart))
return nil
}
//struct upgradeState
func populateDirectCache(db ethdb.Database, blockchain *core.BlockChain) error {
return nil
}

177
trie/directcache.go
View file

@ -17,8 +17,10 @@
package trie
import (
"sync"
"fmt"
"errors"
"time"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/logger"
@ -27,38 +29,15 @@ import (
"github.com/ethereum/go-ethereum/rlp"
)
var directCacheLock = &sync.Mutex{}
var directCacheWrites = metrics.NewCounter("directcache/writes")
var directCacheHitTimer = metrics.NewTimer("directcache/timer/hits")
var directCacheMissTimer = metrics.NewTimer("directcache/timer/misses")
// DirectCacheReads retrieves a global counter measuring the number of direct
// cache reads from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheReads() int64 {
return directCacheHitTimer.Count() + directCacheMissTimer.Count()
}
// DirectCacheWrites retrieves a global counter measuring the number of direct
// cache writes from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheWrites() int64 {
return directCacheWrites.Count()
}
// DirectCacheMisses retrieves a global counter measuring the number of direct
// cache writes from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheMisses() int64 {
return directCacheMissTimer.Count()
}
type cachedValue struct {
Value []byte
BlockNum uint64
BlockHash common.Hash
}
var (
directCacheLock = &sync.Mutex{}
directCacheLocked = false
directCacheWrites = metrics.NewCounter("directcache/writes")
directCacheHits = metrics.NewCounter("directcache/hits")
directCacheMisses = metrics.NewCounter("directcache/misses")
directCacheTimer = metrics.NewTimer("directcache/timer")
NotFound = errors.New("Cache entry not found")
)
// CacheValidator can check whether a certain block is in the current canonical chain.
type CacheValidator interface {
@ -100,10 +79,6 @@ func (dc *DirectCache) Iterator() *Iterator {
return dc.data.Iterator()
}
func (dc *DirectCache) makeKey(key []byte) []byte {
return append(dc.keyPrefix, key...)
}
func (dc *DirectCache) Get(key []byte) []byte {
res, err := dc.TryGet(key)
if err != nil && glog.V(logger.Error) {
@ -113,15 +88,11 @@ func (dc *DirectCache) Get(key []byte) []byte {
}
func (dc *DirectCache) TryGet(key []byte) ([]byte, error) {
start := time.Now()
dirty := dc.dirty[string(key)]
// Use the underlying object for dirty keys
if !dirty {
cacheKey := dc.makeKey(key)
if cached, ok := dc.getCached(cacheKey); ok {
directCacheHitTimer.UpdateSince(start)
if cached, ok := dc.getCached(key); ok {
return cached, nil
}
}
@ -138,26 +109,24 @@ func (dc *DirectCache) TryGet(key []byte) ([]byte, error) {
// Don't count fetches of dirty data as cache misses
if !dirty {
directCacheMissTimer.UpdateSince(start)
directCacheMisses.Inc(1)
}
return value, nil
}
func (dc *DirectCache) getCached(key []byte) ([]byte, bool) {
enc, _ := dc.db.Get(key)
if len(enc) == 0 {
data, blockNum, blockHash, err := GetDirectCache(dc.keyPrefix, key, dc.db)
if err != nil {
if err == NotFound {
return nil, dc.complete
}
var data cachedValue
if err := rlp.DecodeBytes(enc, &data); err != nil && glog.V(logger.Error) {
glog.Errorf("Can't decode cached object at %x: %v", key, err)
glog.Errorf("Error retrieving direct cache data: %v", err)
return nil, false
}
canonical := dc.blockNum > 0 && data.BlockNum < dc.blockNum && dc.validator.IsCanonChainBlock(data.BlockNum, data.BlockHash)
return data.Value, canonical
canonical := dc.blockNum > 0 && blockNum < dc.blockNum && dc.validator.IsCanonChainBlock(blockNum, blockHash)
return data, canonical
}
func (dc *DirectCache) Update(key, value []byte) {
@ -183,19 +152,21 @@ func (dc *DirectCache) TryDelete(key []byte) error {
}
func (dc *DirectCache) CommitTo(dbw DatabaseWriter) (root common.Hash, err error) {
directCacheWrites.Inc(len(dc.dirty))
directCacheLock.Lock()
defer directCacheLock.Unlock()
if err := DirectCacheTransaction(func() error {
for k, _ := range dc.dirty {
v, err := dc.data.TryGet([]byte(k))
if err, ok := err.(*MissingNodeError); err != nil && !ok {
if _, ok := err.(*MissingNodeError); err != nil && !ok {
return err
}
if err := WriteDirectCache(dc.keyPrefix, []byte(k), v, dc.blockNum, dc.blockHash, dbw); err != nil {
return err
}
}
return nil
}); err != nil {
return common.Hash{}, err
}
if err := writeCacheEntry(dc.keyPrefix, []byte(k), v, dc.blockNum, dc.blockHash, dbw); err != nil {
return common.Hash{}, err
}
}
dc.dirty = make(map[string]bool)
return dc.data.CommitTo(dbw)
}
@ -203,20 +174,22 @@ func (dc *DirectCache) CommitTo(dbw DatabaseWriter) (root common.Hash, err error
// Populate iterates over the underlying trie, filling in any unset cache entries.
// After Populate has completed, future DirectCache instances can have `complete`
// set to true, for better efficiency on cache misses.
func (dc *DirectCache) Populate() error {
func (dc *DirectCache) Populate() (err error) {
i := 0
writes := 0
it := dc.Iterator()
for it.Next() {
directCacheLock.Lock()
if val, err := dc.db.Get(append(dc.keyPrefix, it.Key...)); err != nil || val == nil {
if err := DirectCacheTransaction(func() error {
if HasDirectCache(dc.keyPrefix, it.Key, dc.db) {
writes += 1
if err := writeCacheEntry(dc.keyPrefix, it.Key, it.Value, dc.blockNum, dc.blockHash, dc.db); err != nil {
directCacheLock.Unlock()
if err = WriteDirectCache(dc.keyPrefix, it.Key, it.Value, dc.blockNum, dc.blockHash, dc.db); err != nil {
return err
}
}
directCacheLock.Unlock()
return nil
}); err != nil {
return err
}
i += 1
if i % 10000 == 0 && glog.V(logger.Info) {
@ -226,6 +199,26 @@ func (dc *DirectCache) Populate() error {
return nil
}
type cachedValue struct {
Value []byte
BlockNum uint64
BlockHash common.Hash
}
func DirectCacheTransaction(tx func() (error)) error {
if directCacheLocked {
return fmt.Errorf("Reentrant calls to DirectCacheTransaction are not permitted")
}
directCacheLock.Lock()
directCacheLocked = true
defer func() {
directCacheLocked = false
directCacheLock.Unlock()
}()
return tx()
}
// WriteDirectCache places a value node directly into the database along with
// block metadata to validate its relevancy.
@ -234,12 +227,11 @@ func (dc *DirectCache) Populate() error {
// and its integrated cache, namely during fast sync and database upgrades.
func WriteDirectCache(prefix, key, value []byte, number uint64, hash common.Hash, dbw DatabaseWriter) error {
directCacheWrites.Inc(1)
directCacheLock.Lock()
defer directCacheLock.Unlock()
return writeCacheEntry(prefix, key, value, number, hash, dbw)
if !directCacheLocked {
return fmt.Errorf("WriteDirectCache may only be called in a DirectCacheTransaction")
}
func writeCacheEntry(prefix, key, value []byte, number uint64, hash common.Hash, dbw DatabaseWriter) error {
enc, _ := rlp.EncodeToBytes(cachedValue{value, number, hash})
return dbw.Put(append(prefix, key...), enc)
}
@ -249,7 +241,46 @@ func writeCacheEntry(prefix, key, value []byte, number uint64, hash common.Hash,
//
// The method is meant to be used by code that circumvents the state database
// and its integrated cache, namely during fast sync and database upgrades.
func GetDirectCache(prefix, key []byte, db Database) ([]byte, error) {
defer func(start time.Time) { directCacheHitTimer.UpdateSince(start) }(time.Now())
return db.Get(append(prefix, key...))
func GetDirectCache(prefix, key []byte, db Database) ([]byte, uint64, common.Hash, error) {
defer func(start time.Time) { directCacheTimer.UpdateSince(start) }(time.Now())
enc, _ := db.Get(append(prefix, key...))
if len(enc) == 0 {
return nil, 0, common.Hash{}, NotFound
}
var data cachedValue
if err := rlp.DecodeBytes(enc, &data); err != nil && glog.V(logger.Error) {
return nil, 0, common.Hash{}, fmt.Errorf("Can't decode cached object at %x: %v", key, err)
}
return data.Value, data.BlockNum, data.BlockHash, nil
}
// HasDirectCache returns true iff a direct cache node exists for the specified key
func HasDirectCache(prefix, key []byte, db Database) bool {
if enc, err := db.Get(append(prefix, key...)); err == nil && len(enc) > 0 {
return true
}
return false
}
// DirectCacheReads retrieves a global counter measuring the number of direct
// cache reads from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheReads() int64 {
return directCacheTimer.Count()
}
// DirectCacheWrites retrieves a global counter measuring the number of direct
// cache writes from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheWrites() int64 {
return directCacheWrites.Count()
}
// DirectCacheMisses retrieves a global counter measuring the number of direct
// cache misses from the disk since process startup. This isn't useful for anything
// apart from trie debugging purposes.
func DirectCacheMisses() int64 {
return directCacheMisses.Count()
}

2
trie/sync.go
View file

@ -292,7 +292,7 @@ func (s *TrieSync) children(req *request, object node) ([]*request, error) {
// If this branch was already visited, abort iteration
keys := req.keys(append(child.path, it.Nibbles...))
if !unknown {
if val, err := GetDirectCache(req.dcPrefix, keys[0], s.database); val != nil && err == nil {
if HasDirectCache(req.dcPrefix, keys[0], s.database) {
break
}
// Branch is unknown, mark as so to avoid repeated db lookups