pipeline commit trie

core/block_validator.go

@@ -106,7 +106,7 @@ func (v *BlockValidator) ValidateBody(block *types.Block) error {
 // transition, such as amount of used gas, the receipt roots and the state root
 // itself. ValidateState returns a database batch if the validation was a success
 // otherwise nil and an error is returned.
-func (v *BlockValidator) ValidateState(block *types.Block, statedb *state.StateDB, receipts types.Receipts, usedGas uint64) error {
+func (v *BlockValidator) ValidateState(block *types.Block, statedb *state.StateDB, receipts types.Receipts, usedGas uint64, skipHeavyVerify bool) error {
  header := block.Header()
  if block.GasUsed() != usedGas {
  return fmt.Errorf("invalid gas used (remote: %d local: %d)", block.GasUsed(), usedGas)
@@ -129,13 +129,15 @@ func (v *BlockValidator) ValidateState(block *types.Block, statedb *state.StateD
  return nil
  }
  },
- func() error {
+ }
+ if !skipHeavyVerify {
+ validateFuns = append(validateFuns, func() error {
  if root := statedb.IntermediateRoot(v.config.IsEIP158(header.Number)); header.Root != root {
  return fmt.Errorf("invalid merkle root (remote: %x local: %x)", header.Root, root)
  } else {
  return nil
  }
- },
+ })
  }
  validateRes := make(chan error, len(validateFuns))
  for _, f := range validateFuns {

core/blockchain.go

@@ -176,10 +176,11 @@ type BlockChain struct {
  chainConfig *params.ChainConfig // Chain & network configuration
  cacheConfig *CacheConfig // Cache configuration for pruning
 
- db ethdb.Database // Low level persistent database to store final content in
- snaps *snapshot.Tree // Snapshot tree for fast trie leaf access
- triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
- gcproc time.Duration // Accumulates canonical block processing for trie dumping
+ db ethdb.Database // Low level persistent database to store final content in
+ snaps *snapshot.Tree // Snapshot tree for fast trie leaf access
+ triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
+ gcproc time.Duration // Accumulates canonical block processing for trie dumping
+ commitLock sync.Mutex // CommitLock is used to protect above field from being modified concurrently
 
  // txLookupLimit is the maximum number of blocks from head whose tx indices
  // are reserved:
@@ -1049,6 +1050,12 @@ func (bc *BlockChain) HasFastBlock(hash common.Hash, number uint64) bool {
 
 // HasState checks if state trie is fully present in the database or not.
 func (bc *BlockChain) HasState(hash common.Hash) bool {
+ if bc.snaps != nil {
+ // If parent snap is pending on verified, treat it as state exist
+ if s := bc.snaps.Snapshot(hash); s != nil && !s.Verified() {
+ return true
+ }
+ }
  _, err := bc.stateCache.OpenTrie(hash)
  return err == nil
 }
@@ -1660,8 +1667,78 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
  }
  wg.Done()
  }()
+
+ tryCommitTrieDB := func() error {
+ bc.commitLock.Lock()
+ defer bc.commitLock.Unlock()
+
+ triedb := bc.stateCache.TrieDB()
+ // If we're running an archive node, always flush
+ if bc.cacheConfig.TrieDirtyDisabled {
+ err := triedb.Commit(block.Root(), false, nil)
+ if err != nil {
+ return err
+ }
+ } else {
+ // Full but not archive node, do proper garbage collection
+ triedb.Reference(block.Root(), common.Hash{}) // metadata reference to keep trie alive
+ bc.triegc.Push(block.Root(), -int64(block.NumberU64()))
+
+ if current := block.NumberU64(); current > bc.triesInMemory {
+ // If we exceeded our memory allowance, flush matured singleton nodes to disk
+ var (
+ nodes, imgs = triedb.Size()
+ limit = common.StorageSize(bc.cacheConfig.TrieDirtyLimit) * 1024 * 1024
+ )
+ if nodes > limit || imgs > 4*1024*1024 {
+ triedb.Cap(limit - ethdb.IdealBatchSize)
+ }
+ // Find the next state trie we need to commit
+ chosen := current - bc.triesInMemory
+
+ // If we exceeded out time allowance, flush an entire trie to disk
+ if bc.gcproc > bc.cacheConfig.TrieTimeLimit {
+ canWrite := true
+ if posa, ok := bc.engine.(consensus.PoSA); ok {
+ if !posa.EnoughDistance(bc, block.Header()) {
+ canWrite = false
+ }
+ }
+ if canWrite {
+ // If the header is missing (canonical chain behind), we're reorging a low
+ // diff sidechain. Suspend committing until this operation is completed.
+ header := bc.GetHeaderByNumber(chosen)
+ if header == nil {
+ log.Warn("Reorg in progress, trie commit postponed", "number", chosen)
+ } else {
+ // If we're exceeding limits but haven't reached a large enough memory gap,
+ // warn the user that the system is becoming unstable.
+ if chosen < lastWrite+bc.triesInMemory && bc.gcproc >= 2*bc.cacheConfig.TrieTimeLimit {
+ log.Info("State in memory for too long, committing", "time", bc.gcproc, "allowance", bc.cacheConfig.TrieTimeLimit, "optimum", float64(chosen-lastWrite)/float64(bc.triesInMemory))
+ }
+ // Flush an entire trie and restart the counters
+ triedb.Commit(header.Root, true, nil)
+ lastWrite = chosen
+ bc.gcproc = 0
+ }
+ }
+ }
+ // Garbage collect anything below our required write retention
+ for !bc.triegc.Empty() {
+ root, number := bc.triegc.Pop()
+ if uint64(-number) > chosen {
+ bc.triegc.Push(root, number)
+ break
+ }
+ go triedb.Dereference(root.(common.Hash))
+ }
+ }
+ }
+ return nil
+ }
+
  // Commit all cached state changes into underlying memory database.
- root, diffLayer, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()))
+ _, diffLayer, err := state.Commit(bc.chainConfig.IsEIP158(block.Number()), tryCommitTrieDB)
  if err != nil {
  return NonStatTy, err
  }
@@ -1674,69 +1751,9 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
  diffLayer.Number = block.NumberU64()
  bc.cacheDiffLayer(diffLayer)
  }
- triedb := bc.stateCache.TrieDB()
 
- // If we're running an archive node, always flush
- if bc.cacheConfig.TrieDirtyDisabled {
- if err := triedb.Commit(root, false, nil); err != nil {
- return NonStatTy, err
- }
- } else {
- // Full but not archive node, do proper garbage collection
- triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive
- bc.triegc.Push(root, -int64(block.NumberU64()))
-
- if current := block.NumberU64(); current > bc.triesInMemory {
- // If we exceeded our memory allowance, flush matured singleton nodes to disk
- var (
- nodes, imgs = triedb.Size()
- limit = common.StorageSize(bc.cacheConfig.TrieDirtyLimit) * 1024 * 1024
- )
- if nodes > limit || imgs > 4*1024*1024 {
- triedb.Cap(limit - ethdb.IdealBatchSize)
- }
- // Find the next state trie we need to commit
- chosen := current - bc.triesInMemory
-
- // If we exceeded out time allowance, flush an entire trie to disk
- if bc.gcproc > bc.cacheConfig.TrieTimeLimit {
- canWrite := true
- if posa, ok := bc.engine.(consensus.PoSA); ok {
- if !posa.EnoughDistance(bc, block.Header()) {
- canWrite = false
- }
- }
- if canWrite {
- // If the header is missing (canonical chain behind), we're reorging a low
- // diff sidechain. Suspend committing until this operation is completed.
- header := bc.GetHeaderByNumber(chosen)
- if header == nil {
- log.Warn("Reorg in progress, trie commit postponed", "number", chosen)
- } else {
- // If we're exceeding limits but haven't reached a large enough memory gap,
- // warn the user that the system is becoming unstable.
- if chosen < lastWrite+bc.triesInMemory && bc.gcproc >= 2*bc.cacheConfig.TrieTimeLimit {
- log.Info("State in memory for too long, committing", "time", bc.gcproc, "allowance", bc.cacheConfig.TrieTimeLimit, "optimum", float64(chosen-lastWrite)/float64(bc.triesInMemory))
- }
- // Flush an entire trie and restart the counters
- triedb.Commit(header.Root, true, nil)
- lastWrite = chosen
- bc.gcproc = 0
- }
- }
- }
- // Garbage collect anything below our required write retention
- for !bc.triegc.Empty() {
- root, number := bc.triegc.Pop()
- if uint64(-number) > chosen {
- bc.triegc.Push(root, number)
- break
- }
- go triedb.Dereference(root.(common.Hash))
- }
- }
- }
  wg.Wait()
+
  // If the total difficulty is higher than our known, add it to the canonical chain
  // Second clause in the if statement reduces the vulnerability to selfish mining.
  // Please refer to http://www.cs.cornell.edu/~ie53/publications/btcProcFC.pdf
@@ -2054,6 +2071,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
 
  //Process block using the parent state as reference point
  substart := time.Now()
+ statedb.SetExpectedStateRoot(block.Root())
  statedb, receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig)
  activeState = statedb
  if err != nil {
@@ -2073,7 +2091,8 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, er
  // Validate the state using the default validator
  substart = time.Now()
  if !statedb.IsLightProcessed() {
- if err := bc.validator.ValidateState(block, statedb, receipts, usedGas); err != nil {
+ // Do the state root verification asynchronously
+ if err := bc.validator.ValidateState(block, statedb, receipts, usedGas, true); err != nil {
  log.Error("validate state failed", "error", err)
  bc.reportBlock(block, receipts, err)
  return it.index, err

core/blockchain_diff_test.go

@@ -317,6 +317,9 @@ func TestProcessDiffLayer(t *testing.T) {
  lightBackend.Chain().HandleDiffLayer(diff, "testpid", true)
  }
  _, err := lightBackend.chain.insertChain([]*types.Block{block}, true)
+ if err != nil {
+ t.Errorf("failed to insert block %v", err)
+ }
  if checks, exist := checkBlocks[i]; exist {
  for _, check := range checks.txs {
  s, _ := lightBackend.Chain().Snapshots().Snapshot(block.Root()).Storage(crypto.Keccak256Hash((*check.to)[:]), check.slot)
@@ -325,9 +328,6 @@ func TestProcessDiffLayer(t *testing.T) {
  }
  }
  }
- if err != nil {
- t.Errorf("failed to insert block %v", err)
- }
  }
  currentBlock := lightBackend.chain.CurrentBlock()
  nextBlock := fullBackend.chain.GetBlockByNumber(currentBlock.NumberU64() + 1)
@@ -368,7 +368,7 @@ func TestFreezeDiffLayer(t *testing.T) {
  // Wait for the buffer to be zero.
  }
  // Minus one empty block.
- if fullBackend.chain.diffQueue.Size() != blockNum-1 {
+ if fullBackend.chain.diffQueue.Size() > blockNum-1 && fullBackend.chain.diffQueue.Size() < blockNum-2 {
  t.Errorf("size of diff queue is wrong, expected: %d, get: %d", blockNum-1, fullBackend.chain.diffQueue.Size())
  }
 

core/blockchain_test.go

@@ -151,12 +151,13 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error {
  if err != nil {
  return err
  }
+ statedb.SetExpectedStateRoot(block.Root())
  statedb, receipts, _, usedGas, err := blockchain.processor.Process(block, statedb, vm.Config{})
  if err != nil {
  blockchain.reportBlock(block, receipts, err)
  return err
  }
- err = blockchain.validator.ValidateState(block, statedb, receipts, usedGas)
+ err = blockchain.validator.ValidateState(block, statedb, receipts, usedGas, false)
  if err != nil {
  blockchain.reportBlock(block, receipts, err)
  return err

core/state/database.go

@@ -257,6 +257,9 @@ func (db *cachingDB) Purge() {
 
 // CopyTrie returns an independent copy of the given trie.
 func (db *cachingDB) CopyTrie(t Trie) Trie {
+ if t == nil {
+ return nil
+ }
  switch t := t.(type) {
  case *trie.SecureTrie:
  return t.Copy()

core/state/snapshot/difflayer.go

@@ -118,6 +118,10 @@ type diffLayer struct {
  storageList map[common.Hash][]common.Hash // List of storage slots for iterated retrievals, one per account. Any existing lists are sorted if non-nil
  storageData map[common.Hash]map[common.Hash][]byte // Keyed storage slots for direct retrieval. one per account (nil means deleted)
 
+ // the difflayer is verified when verifiedCh is nil or closed
+ verifiedCh chan struct{}
+ verifyRes bool
+
  diffed *bloomfilter.Filter // Bloom filter tracking all the diffed items up to the disk layer
 
  lock sync.RWMutex
@@ -168,7 +172,7 @@ func (h storageBloomHasher) Sum64() uint64 {
 
 // newDiffLayer creates a new diff on top of an existing snapshot, whether that's a low
 // level persistent database or a hierarchical diff already.
-func newDiffLayer(parent snapshot, root common.Hash, destructs map[common.Hash]struct{}, accounts map[common.Hash][]byte, storage map[common.Hash]map[common.Hash][]byte) *diffLayer {
+func newDiffLayer(parent snapshot, root common.Hash, destructs map[common.Hash]struct{}, accounts map[common.Hash][]byte, storage map[common.Hash]map[common.Hash][]byte, verified chan struct{}) *diffLayer {
  // Create the new layer with some pre-allocated data segments
  dl := &diffLayer{
  parent: parent,
@@ -177,6 +181,7 @@ func newDiffLayer(parent snapshot, root common.Hash, destructs map[common.Hash]s
  accountData: accounts,
  storageData: storage,
  storageList: make(map[common.Hash][]common.Hash),
+ verifiedCh: verified,
  }
  switch parent := parent.(type) {
  case *diskLayer:
@@ -256,6 +261,31 @@ func (dl *diffLayer) Root() common.Hash {
  return dl.root
 }
 
+// WaitVerified will wait until the diff layer been verified
+func (dl *diffLayer) WaitVerified() bool {
+ if dl.verifiedCh == nil {
+ return true
+ }
+ <-dl.verifiedCh
+ return dl.verifyRes
+}
+
+func (dl *diffLayer) MarkVerified() {
+ dl.verifyRes = true
+}
+
+func (dl *diffLayer) Verified() bool {
+ if dl.verifiedCh == nil {
+ return true
+ }
+ select {
+ case <-dl.verifiedCh:
+ return true
+ default:
+ return false
+ }
+}
+
 // Parent returns the subsequent layer of a diff layer.
 func (dl *diffLayer) Parent() snapshot {
  return dl.parent
@@ -423,8 +453,8 @@ func (dl *diffLayer) storage(accountHash, storageHash common.Hash, depth int) ([
 
 // Update creates a new layer on top of the existing snapshot diff tree with
 // the specified data items.
-func (dl *diffLayer) Update(blockRoot common.Hash, destructs map[common.Hash]struct{}, accounts map[common.Hash][]byte, storage map[common.Hash]map[common.Hash][]byte) *diffLayer {
- return newDiffLayer(dl, blockRoot, destructs, accounts, storage)
+func (dl *diffLayer) Update(blockRoot common.Hash, destructs map[common.Hash]struct{}, accounts map[common.Hash][]byte, storage map[common.Hash]map[common.Hash][]byte, verified chan struct{}) *diffLayer {
+ return newDiffLayer(dl, blockRoot, destructs, accounts, storage, verified)
 }
 
 // flatten pushes all data from this point downwards, flattening everything into