From f40e3442a1acd778e68c1c15056459d1258b3214 Mon Sep 17 00:00:00 2001 From: Nick Johnson Date: Wed, 26 Oct 2016 11:07:35 +0100 Subject: [PATCH] core, eth, miner, trie: miner optimizations and database migration --- core/blockchain.go | 3 +- core/chain_makers.go | 2 +- core/state/iterator_test.go | 2 +- core/state/statedb.go | 31 +++-- core/state/sync.go | 13 ++- core/state_processor.go | 2 +- eth/backend.go | 4 + eth/db_upgrade.go | 35 ++++++ miner/worker.go | 2 +- trie/directcache.go | 224 ++++++++++++++++++++++++++---------- trie/sync.go | 2 +- 11 files changed, 239 insertions(+), 81 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 5de3bacd6f..288b5d8bbe 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -553,7 +553,7 @@ func (self *BlockChain) GetBlockByNumber(number uint64) *types.Block { // IsCanonChainBlock checks whether the given block is in the current canonical chain. func (self *BlockChain) IsCanonChainBlock(number uint64, hash common.Hash) bool { - return number < uint64(self.currentBlock.Number().Int64()) && GetCanonicalHash(self.chainDb, number) == hash + return GetCanonicalHash(self.chainDb, number) == hash } // [deprecated by eth/62] @@ -937,6 +937,7 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) { self.reportBlock(block, nil, err) return i, err } + self.stateCache.SetBlockContext(block.Hash(), block.NumberU64(), self) // Process block using the parent state as reference point. receipts, logs, usedGas, err := self.processor.Process(block, self.stateCache, vm.Config{}) if err != nil { diff --git a/core/chain_makers.go b/core/chain_makers.go index 2170b10656..2d115f48bc 100644 --- a/core/chain_makers.go +++ b/core/chain_makers.go @@ -107,7 +107,7 @@ func (b *BlockGen) AddTx(tx *types.Transaction) { if b.gasPool == nil { b.SetCoinbase(common.Address{}) } - b.statedb.SetTxContext(common.Hash{}, b.header.Number.Uint64(), tx.Hash(), len(b.txs), nil) + b.statedb.SetTxContext(tx.Hash(), len(b.txs)) receipt, _, _, err := ApplyTransaction(b.config, nil, b.gasPool, b.statedb, b.header, tx, b.header.GasUsed, vm.Config{}) if err != nil { panic(err) 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 40fbb6a037..94e2bc2af2 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -37,7 +37,7 @@ import ( // Trie cache generation limit after which to evic trie nodes from memory. var MaxTrieCacheGen = uint16(120) -var CachePrefix = []byte("accounthashcache:") +var DirectCachePrefix = []byte("accounthashcache:") const ( // Number of past tries to keep. This value is chosen such that @@ -62,6 +62,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 @@ -103,7 +104,7 @@ func New(root common.Hash, db ethdb.Database) (*StateDB, error) { refund: new(big.Int), logs: make(map[common.Hash]vm.Logs), } - ret.SetTxContext(common.Hash{}, 0, common.Hash{}, 0, nil) + ret.SetBlockContext(common.Hash{}, 0, nil) return ret, nil } @@ -126,7 +127,7 @@ func (self *StateDB) New(root common.Hash) (*StateDB, error) { refund: new(big.Int), logs: make(map[common.Hash]vm.Logs), } - ret.SetTxContext(common.Hash{}, 0, common.Hash{}, 0, nil) + ret.SetBlockContext(common.Hash{}, 0, nil) return ret, nil } @@ -149,8 +150,8 @@ func (self *StateDB) Reset(root common.Hash) error { self.logs = make(map[common.Hash]vm.Logs) self.logSize = 0 self.clearJournalAndRefund() - self.SetTxContext(common.Hash{}, 0, common.Hash{}, 0, nil) - + self.SetBlockContext(common.Hash{}, 0, nil) + self.SetTxContext(common.Hash{}, 0) return nil } @@ -179,18 +180,28 @@ func (self *StateDB) pushTrie(t *trie.Trie) { } } -func (self *StateDB) SetTxContext(blockHash common.Hash, blockNum uint64, txHash common.Hash, txIndex int, validator trie.CacheValidator) { +func (self *StateDB) SetBlockContext(blockHash common.Hash, blockNum uint64, validator trie.CacheValidator) { self.bhash = blockHash - self.thash = txHash - self.txIndex = txIndex - if validator == nil { validator = &trie.NullCacheValidator{} } - storage := trie.NewDirectCache(self.trie, self.db, CachePrefix, validator, true) + + // Check if cache population has completed + if !self.cacheComplete { + if _, status := trie.GetMigrationState(DirectCachePrefix, self.db); status == trie.Complete { + self.cacheComplete = true + } + } + + storage := trie.NewDirectCache(self.trie, self.db, DirectCachePrefix, blockNum, blockHash, validator, self.cacheComplete) self.storage = trie.NewSecure(storage, self.db) } +func (self *StateDB) SetTxContext(txHash common.Hash, txIndex int) { + self.thash = txHash + self.txIndex = txIndex +} + func (self *StateDB) AddLog(log *vm.Log) { self.journal = append(self.journal, addLogChange{txhash: self.thash}) diff --git a/core/state/sync.go b/core/state/sync.go index 1f69a9ef1b..613ae3715f 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/core/state_processor.go b/core/state_processor.go index e6c0401ad1..17e0b539ff 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -72,7 +72,7 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg } // Iterate over and process the individual transactions for i, tx := range block.Transactions() { - statedb.SetTxContext(block.Hash(), block.NumberU64(), tx.Hash(), i, p.bc) + statedb.SetTxContext(tx.Hash(), i) receipt, logs, _, err := ApplyTransaction(p.config, p.bc, gp, statedb, header, tx, totalUsedGas, cfg) if err != nil { return nil, nil, nil, err diff --git a/eth/backend.go b/eth/backend.go index d5b767b125..bc4818b4db 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -255,6 +255,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 5fd73a5867..ecf8c4dd95 100644 --- a/eth/db_upgrade.go +++ b/eth/db_upgrade.go @@ -26,11 +26,13 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" ) var useSequentialKeys = []byte("dbUpgrade_20160530sequentialKeys") @@ -354,3 +356,36 @@ func addMipmapBloomBins(db ethdb.Database) (err error) { glog.V(logger.Info).Infoln("upgrade completed in", time.Since(tstart)) return nil } + +func populateDirectCache(db ethdb.Database, bc *core.BlockChain) error { + blockHash, status := trie.GetMigrationState(state.DirectCachePrefix, db) + if status == trie.NotStarted { + blockHash = core.GetHeadBlockHash(db) + status = trie.Running + trie.SetMigrationState(state.DirectCachePrefix, blockHash, status, db) + } + blockNum := core.GetBlockNumber(db, blockHash) + + if status == trie.Running { + tr, err := trie.New(core.GetBlock(db, blockHash, blockNum).Root(), db, 0) + if err != nil { + return err + } + dc := trie.NewDirectCache(tr, db, state.DirectCachePrefix, blockNum, blockHash, bc, false) + go func() { + for core.GetBlockNumber(db, core.GetHeadBlockHash(db)) < blockNum+8 { + time.Sleep(10 * time.Second) + } + glog.Infof("Beginning direct cache upgrade at block %v", blockNum) + + start := time.Now() + if err := dc.Populate(); err != nil { + glog.Errorf("Direct cache upgrade failed: %v", err) + } else { + glog.Infof("Direct cache upgrade finished in %v", time.Since(start)) + } + trie.SetMigrationState(state.DirectCachePrefix, blockHash, trie.Complete, db) + }() + } + return nil +} diff --git a/miner/worker.go b/miner/worker.go index 175851e4c2..b72bc2f6da 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -617,7 +617,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB continue } // Start executing the transaction - env.state.SetTxContext(common.Hash{}, env.header.Number.Uint64(), tx.Hash(), env.tcount, bc) + env.state.SetTxContext(tx.Hash(), env.tcount) err, logs := env.commitTransaction(tx, bc, gp) switch { diff --git a/trie/directcache.go b/trie/directcache.go index 1ca04f2284..c7e114c1bc 100644 --- a/trie/directcache.go +++ b/trie/directcache.go @@ -17,7 +17,10 @@ package trie import ( + "fmt" + "errors" "time" + "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/logger" @@ -26,36 +29,24 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) -var directCacheWrites = metrics.NewCounter("directcache/writes") -var directCacheHitTimer = metrics.NewTimer("directcache/timer/hits") -var directCacheMissTimer = metrics.NewTimer("directcache/timer/misses") +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") + MigrationPrefix = []byte("directstatecachemigration:") +) -// 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() -} +type MigrationStatus int -// 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 -} +const ( + NotStarted = iota + Running = iota + Complete = iota +) // CacheValidator can check whether a certain block is in the current canonical chain. type CacheValidator interface { @@ -79,11 +70,13 @@ func (cv *NullCacheValidator) IsCanonChainBlock(num uint64, hash common.Hash) bo return false } -func NewDirectCache(pm PersistentMap, db Database, keyPrefix []byte, validator CacheValidator, complete bool) *DirectCache { +func NewDirectCache(pm PersistentMap, db Database, keyPrefix []byte, blockNum uint64, blockHash common.Hash, validator CacheValidator, complete bool) *DirectCache { return &DirectCache{ data: pm, db: db, keyPrefix: keyPrefix, + blockNum: blockNum, + blockHash: blockHash, validator: validator, complete: complete, dirty: make(map[string]bool), @@ -95,10 +88,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) { @@ -108,15 +97,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 } } @@ -133,26 +118,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.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) { @@ -178,21 +161,68 @@ func (dc *DirectCache) TryDelete(key []byte) error { } func (dc *DirectCache) CommitTo(dbw DatabaseWriter) (root common.Hash, err error) { - 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 := dc.putCache(dbw, []byte(k), v); 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) } -func (dc *DirectCache) putCache(dbw DatabaseWriter, key, value []byte) error { - return WriteDirectCache(dc.keyPrefix, key, value, dc.blockNum, dc.blockHash, dbw) +// 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() (err error) { + i := 0 + writes := 0 + it := dc.Iterator() + for it.Next() { + 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 + } + + i += 1 + if i % 10000 == 0 && glog.V(logger.Info) { + glog.V(logger.Info).Infof("Constructing direct cache: processed %v entries, written %v", i, writes) + } + } + return nil +} + +type cachedValue struct { + Value []byte + BlockNum uint64 + BlockHash common.Hash +} + +func DirectCacheTransaction(tx func() (error)) error { + directCacheLock.Lock() + directCacheLocked = true + defer func() { + directCacheLocked = false + directCacheLock.Unlock() + }() + + return tx() } // WriteDirectCache places a value node directly into the database along with @@ -202,6 +232,11 @@ func (dc *DirectCache) putCache(dbw DatabaseWriter, key, value []byte) 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) + + 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) } @@ -211,7 +246,74 @@ func WriteDirectCache(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 { + 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 +} + +type migrationState struct { + Status MigrationStatus + Hash common.Hash +} + +// GetMigrationState returns the block number of the migration to the direct cache, and +// whether or not it's complete. +func GetMigrationState(prefix []byte, db Database) (common.Hash, MigrationStatus) { + enc, _ := db.Get(append(MigrationPrefix, prefix...)) + if len(enc) == 0 { + return common.Hash{}, NotStarted + } + + var data migrationState + if err := rlp.DecodeBytes(enc, &data); err != nil { + glog.Errorf("Could not decode migration status: %v", err) + return common.Hash{}, NotStarted + } + + return data.Hash, data.Status +} + +// SetMigrationState updates the migration state in the database +func SetMigrationState(prefix []byte, blockHash common.Hash, status MigrationStatus, db Database) error { + enc, _ := rlp.EncodeToBytes(migrationState{status, blockHash}) + return db.Put(append(MigrationPrefix, prefix...), enc) +} + +// 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 2d0d49ca97..8a91ddf293 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -298,7 +298,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