core, eth, miner, trie: miner optimizations and database migration

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 {

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)

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 {

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})
 

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.DirectCacheTransaction(func() error {
-			if err := trie.WriteDirectCache(CachePrefix, key, leaf, number, hash, database); err != nil {
+			for _, key := range keys {
-				return err
+				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)
 }
 

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

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
 }
 

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
+}

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 {

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 (
-var directCacheHitTimer = metrics.NewTimer("directcache/timer/hits")
+	directCacheLock              = &sync.Mutex{}
-var directCacheMissTimer = metrics.NewTimer("directcache/timer/misses")
+	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
+type MigrationStatus int
-// 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
+const (
-// cache writes from the disk since process startup. This isn't useful for anything
+	NotStarted = iota
-// apart from trie debugging purposes.
+	Running    = iota
-func DirectCacheWrites() int64 {
+	Complete   = iota
-	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
-}
 
 // 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(key); ok {
-		if cached, ok := dc.getCached(cacheKey); ok {
-			directCacheHitTimer.UpdateSince(start)
 			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)
+	data, blockNum, blockHash, err := GetDirectCache(dc.keyPrefix, key, dc.db)
-	if len(enc) == 0 {
+	if err != nil {
-		return nil, dc.complete
+		if err == NotFound {
-	}
+			return nil, dc.complete
-
+		}
-	var data cachedValue
+		glog.Errorf("Error retrieving direct cache data: %v", err)
-	if err := rlp.DecodeBytes(enc, &data); err != nil && glog.V(logger.Error) {
-		glog.Errorf("Can't decode cached object at %x: %v", key, err)
 		return nil, false
 	}
 
-	canonical := dc.validator.IsCanonChainBlock(data.BlockNum, data.BlockHash)
+	canonical := dc.blockNum > 0 && blockNum < dc.blockNum && dc.validator.IsCanonChainBlock(blockNum, blockHash)
-	return data.Value, canonical
+	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 {
+	if err := DirectCacheTransaction(func() error {
-		v, err := dc.data.TryGet([]byte(k))
+		for k, _ := range dc.dirty {
-		if err, ok := err.(*MissingNodeError); err != nil && !ok {
+			v, err := dc.data.TryGet([]byte(k))
-			return common.Hash{}, err
+			if _, ok := err.(*MissingNodeError); err != nil && !ok {
-		}
+				return err
-		if err := dc.putCache(dbw, []byte(k), v); err != nil {
+			}
-			return common.Hash{}, 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 {
+// Populate iterates over the underlying trie, filling in any unset cache entries.
-	return WriteDirectCache(dc.keyPrefix, key, value, dc.blockNum, dc.blockHash, dbw)
+// 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) {
+func GetDirectCache(prefix, key []byte, db Database) ([]byte, uint64, common.Hash, error) {
-	defer func(start time.Time) { directCacheHitTimer.UpdateSince(start) }(time.Now())
+	defer func(start time.Time) { directCacheTimer.UpdateSince(start) }(time.Now())
-	return db.Get(append(prefix, key...))
+
+	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()
 }

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