From 32622544967bf93d146ca1135cbb84ee89a3d2a9 Mon Sep 17 00:00:00 2001 From: Nick Johnson Date: Thu, 27 Oct 2016 14:10:11 +0100 Subject: [PATCH] core/state, trie: Refactored direct cache backend to be more coherent. --- core/state/iterator_test.go | 2 +- core/state/statedb.go | 14 ++- core/state/sync.go | 13 ++- eth/backend.go | 4 + eth/db_upgrade.go | 7 ++ trie/directcache.go | 191 +++++++++++++++++++++--------------- trie/sync.go | 2 +- 7 files changed, 145 insertions(+), 88 deletions(-) diff --git a/core/state/iterator_test.go b/core/state/iterator_test.go index dc16523eff..369f2a9998 100644 --- a/core/state/iterator_test.go +++ b/core/state/iterator_test.go @@ -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 { diff --git a/core/state/statedb.go b/core/state/statedb.go index c0c731eda2..fe29e318a5 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -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) } diff --git a/core/state/sync.go b/core/state/sync.go index 28c0fb7731..83cbc90320 100644 --- a/core/state/sync.go +++ b/core/state/sync.go @@ -47,10 +47,15 @@ func NewStateSync(number uint64, hash common.Hash, root common.Hash, database et return err } // Populate the direct account caches in the database - for _, key := range keys { - if err := trie.WriteDirectCache(CachePrefix, key, leaf, number, hash, database); err != nil { - return err + if err := trie.DirectCacheTransaction(func() error { + for _, key := range keys { + 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) @@ -58,7 +63,7 @@ func NewStateSync(number uint64, hash common.Hash, root common.Hash, database et return nil } - syncer = trie.NewTrieSync(root, database, callback, CachePrefix) + syncer = trie.NewTrieSync(root, database, callback, DirectCachePrefix) return (*StateSync)(syncer) } diff --git a/eth/backend.go b/eth/backend.go index c4a883c9e9..7eba71718e 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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 } diff --git a/eth/db_upgrade.go b/eth/db_upgrade.go index 172bb0954a..f7ba391576 100644 --- a/eth/db_upgrade.go +++ b/eth/db_upgrade.go @@ -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 +} diff --git a/trie/directcache.go b/trie/directcache.go index bbb2f6e23c..84b555be13 100644 --- a/trie/directcache.go +++ b/trie/directcache.go @@ -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 { - 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) + data, blockNum, blockHash, err := GetDirectCache(dc.keyPrefix, key, dc.db) + if err != nil { + if err == NotFound { + return nil, dc.complete + } + 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() - - for k, _ := range dc.dirty { - v, err := dc.data.TryGet([]byte(k)) - if err, ok := err.(*MissingNodeError); err != nil && !ok { - return common.Hash{}, err - } - if err := writeCacheEntry(dc.keyPrefix, []byte(k), v, dc.blockNum, dc.blockHash, dbw); err != nil { - return common.Hash{}, err + if err := DirectCacheTransaction(func() error { + for k, _ := range dc.dirty { + v, err := dc.data.TryGet([]byte(k)) + 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 } + 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 { - writes += 1 - if err := writeCacheEntry(dc.keyPrefix, it.Key, it.Value, dc.blockNum, dc.blockHash, dc.db); err != nil { - directCacheLock.Unlock() - return err + if err := DirectCacheTransaction(func() error { + if HasDirectCache(dc.keyPrefix, it.Key, dc.db) { + writes += 1 + if err = WriteDirectCache(dc.keyPrefix, it.Key, it.Value, dc.blockNum, dc.blockHash, dc.db); err != nil { + return err + } } + return nil + }); err != nil { + return err } - directCacheLock.Unlock() 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) -} -func writeCacheEntry(prefix, key, value []byte, number uint64, hash common.Hash, dbw DatabaseWriter) error { + if !directCacheLocked { + return fmt.Errorf("WriteDirectCache may only be called in a DirectCacheTransaction") + } + 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() } diff --git a/trie/sync.go b/trie/sync.go index 8ab789f4ab..335c7a2a0f 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -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