fix concurrent issue for trie and subfetchers

core/state/dump.go

@@ -138,7 +138,7 @@ func (s *StateDB) DumpToCollector(c DumpCollector, excludeCode, excludeStorage,
 			account.SecureKey = it.Key
 		}
 		addr := common.BytesToAddress(addrBytes)
-		obj := newObject(s, addr, data)
+		obj := newObject(s, s.isParallel, addr, data)
 		if !excludeCode {
 			account.Code = common.Bytes2Hex(obj.Code(s.db))
 		}

core/state/interface.go

@@ -0,0 +1,82 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package state
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+// StateDBer is copied from vm/interface.go
+// It is used by StateObject & Journal right now, to abstract StateDB & ParallelStateDB
+type StateDBer interface {
+	getBaseStateDB() *StateDB
+	getStateObject(common.Address) *StateObject // only accessible for journal
+	storeStateObj(common.Address, *StateObject) // only accessible for journal
+
+	CreateAccount(common.Address)
+
+	SubBalance(common.Address, *big.Int)
+	AddBalance(common.Address, *big.Int)
+	GetBalance(common.Address) *big.Int
+
+	GetNonce(common.Address) uint64
+	SetNonce(common.Address, uint64)
+
+	GetCodeHash(common.Address) common.Hash
+	GetCode(common.Address) []byte
+	SetCode(common.Address, []byte)
+	GetCodeSize(common.Address) int
+
+	AddRefund(uint64)
+	SubRefund(uint64)
+	GetRefund() uint64
+
+	GetCommittedState(common.Address, common.Hash) common.Hash
+	GetState(common.Address, common.Hash) common.Hash
+	SetState(common.Address, common.Hash, common.Hash)
+
+	Suicide(common.Address) bool
+	HasSuicided(common.Address) bool
+
+	// Exist reports whether the given account exists in state.
+	// Notably this should also return true for suicided accounts.
+	Exist(common.Address) bool
+	// Empty returns whether the given account is empty. Empty
+	// is defined according to EIP161 (balance = nonce = code = 0).
+	Empty(common.Address) bool
+
+	PrepareAccessList(sender common.Address, dest *common.Address, precompiles []common.Address, txAccesses types.AccessList)
+	AddressInAccessList(addr common.Address) bool
+	SlotInAccessList(addr common.Address, slot common.Hash) (addressOk bool, slotOk bool)
+	// AddAddressToAccessList adds the given address to the access list. This operation is safe to perform
+	// even if the feature/fork is not active yet
+	AddAddressToAccessList(addr common.Address)
+	// AddSlotToAccessList adds the given (address,slot) to the access list. This operation is safe to perform
+	// even if the feature/fork is not active yet
+	AddSlotToAccessList(addr common.Address, slot common.Hash)
+
+	RevertToSnapshot(int)
+	Snapshot() int
+
+	AddLog(*types.Log)
+	AddPreimage(common.Hash, []byte)
+
+	ForEachStorage(common.Address, func(common.Hash, common.Hash) bool) error
+}

core/state/journal.go

@@ -26,7 +26,7 @@ import (
 // reverted on demand.
 type journalEntry interface {
 	// revert undoes the changes introduced by this journal entry.
-	revert(*StateDB)
+	revert(StateDBer)
 
 	// dirtied returns the Ethereum address modified by this journal entry.
 	dirtied() *common.Address
@@ -58,10 +58,10 @@ func (j *journal) append(entry journalEntry) {
 
 // revert undoes a batch of journalled modifications along with any reverted
 // dirty handling too.
-func (j *journal) revert(statedb *StateDB, snapshot int) {
+func (j *journal) revert(dber StateDBer, snapshot int) {
 	for i := len(j.entries) - 1; i >= snapshot; i-- {
 		// Undo the changes made by the operation
-		j.entries[i].revert(statedb)
+		j.entries[i].revert(dber)
 
 		// Drop any dirty tracking induced by the change
 		if addr := j.entries[i].dirtied(); addr != nil {
@@ -141,9 +141,15 @@ type (
 	}
 )
 
-func (ch createObjectChange) revert(s *StateDB) {
+func (ch createObjectChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
 	if s.parallel.isSlotDB {
 		delete(s.parallel.dirtiedStateObjectsInSlot, *ch.account)
+		delete(s.parallel.addrStateChangesInSlot, *ch.account)
+		delete(s.parallel.nonceChangesInSlot, *ch.account)
+		delete(s.parallel.balanceChangesInSlot, *ch.account)
+		delete(s.parallel.codeChangesInSlot, *ch.account)
+		delete(s.parallel.kvChangesInSlot, *ch.account)
 	} else {
 		s.deleteStateObj(*ch.account)
 	}
@@ -154,8 +160,15 @@ func (ch createObjectChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch resetObjectChange) revert(s *StateDB) {
-	s.SetStateObject(ch.prev)
+func (ch resetObjectChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
+	if s.parallel.isSlotDB {
+		// ch.prev must be from dirtiedStateObjectsInSlot, put it back
+		s.parallel.dirtiedStateObjectsInSlot[ch.prev.address] = ch.prev
+	} else {
+		// ch.prev was got from main DB, put it back to main DB.
+		s.storeStateObj(ch.prev.address, ch.prev)
+	}
 	if !ch.prevdestruct && s.snap != nil {
 		delete(s.snapDestructs, ch.prev.address)
 	}
@@ -165,8 +178,8 @@ func (ch resetObjectChange) dirtied() *common.Address {
 	return nil
 }
 
-func (ch suicideChange) revert(s *StateDB) {
-	obj := s.getStateObject(*ch.account)
+func (ch suicideChange) revert(dber StateDBer) {
+	obj := dber.getStateObject(*ch.account)
 	if obj != nil {
 		obj.suicided = ch.prev
 		obj.setBalance(ch.prevbalance)
@@ -179,54 +192,57 @@ func (ch suicideChange) dirtied() *common.Address {
 
 var ripemd = common.HexToAddress("0000000000000000000000000000000000000003")
 
-func (ch touchChange) revert(s *StateDB) {
+func (ch touchChange) revert(dber StateDBer) {
 }
 
 func (ch touchChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch balanceChange) revert(s *StateDB) {
-	s.getStateObject(*ch.account).setBalance(ch.prev)
+func (ch balanceChange) revert(dber StateDBer) {
+	dber.getStateObject(*ch.account).setBalance(ch.prev)
 }
 
 func (ch balanceChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch nonceChange) revert(s *StateDB) {
-	s.getStateObject(*ch.account).setNonce(ch.prev)
+func (ch nonceChange) revert(dber StateDBer) {
+	dber.getStateObject(*ch.account).setNonce(ch.prev)
 }
 
 func (ch nonceChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch codeChange) revert(s *StateDB) {
-	s.getStateObject(*ch.account).setCode(common.BytesToHash(ch.prevhash), ch.prevcode)
+func (ch codeChange) revert(dber StateDBer) {
+	dber.getStateObject(*ch.account).setCode(common.BytesToHash(ch.prevhash), ch.prevcode)
 }
 
 func (ch codeChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch storageChange) revert(s *StateDB) {
-	s.getStateObject(*ch.account).setState(ch.key, ch.prevalue)
+func (ch storageChange) revert(dber StateDBer) {
+	dber.getStateObject(*ch.account).setState(ch.key, ch.prevalue)
 }
 
 func (ch storageChange) dirtied() *common.Address {
 	return ch.account
 }
 
-func (ch refundChange) revert(s *StateDB) {
+func (ch refundChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
 	s.refund = ch.prev
 }
 
 func (ch refundChange) dirtied() *common.Address {
 	return nil
 }
 
-func (ch addLogChange) revert(s *StateDB) {
+func (ch addLogChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
+
 	logs := s.logs[ch.txhash]
 	if len(logs) == 1 {
 		delete(s.logs, ch.txhash)
@@ -240,15 +256,17 @@ func (ch addLogChange) dirtied() *common.Address {
 	return nil
 }
 
-func (ch addPreimageChange) revert(s *StateDB) {
+func (ch addPreimageChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
 	delete(s.preimages, ch.hash)
 }
 
 func (ch addPreimageChange) dirtied() *common.Address {
 	return nil
 }
 
-func (ch accessListAddAccountChange) revert(s *StateDB) {
+func (ch accessListAddAccountChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
 	/*
 		One important invariant here, is that whenever a (addr, slot) is added, if the
 		addr is not already present, the add causes two journal entries:
@@ -267,7 +285,8 @@ func (ch accessListAddAccountChange) dirtied() *common.Address {
 	return nil
 }
 
-func (ch accessListAddSlotChange) revert(s *StateDB) {
+func (ch accessListAddSlotChange) revert(dber StateDBer) {
+	s := dber.getBaseStateDB()
 	if s.accessList != nil {
 		s.accessList.DeleteSlot(*ch.address, *ch.slot)
 	}

core/state/snapshot/difflayer.go

@@ -26,9 +26,10 @@ import (
 	"sync/atomic"
 	"time"
 
+	bloomfilter "github.com/holiman/bloomfilter/v2"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rlp"
-	bloomfilter "github.com/holiman/bloomfilter/v2"
 )
 
 var (
@@ -213,6 +214,10 @@ func newDiffLayer(parent snapshot, root common.Hash, destructs map[common.Hash]s
 	return dl
 }
 
+func (dl *diffLayer) Generating() bool {
+	return dl.origin.Generating()
+}
+
 // rebloom discards the layer's current bloom and rebuilds it from scratch based
 // on the parent's and the local diffs.
 func (dl *diffLayer) rebloom(origin *diskLayer) {

core/state/snapshot/disklayer.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/fastcache"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -49,6 +50,10 @@ func (dl *diskLayer) Root() common.Hash {
 	return dl.root
 }
 
+func (dl *diskLayer) Generating() bool {
+	return dl.stale || (dl.genMarker != nil)
+}
+
 func (dl *diskLayer) WaitAndGetVerifyRes() bool {
 	return true
 }

core/state/snapshot/snapshot.go

@@ -121,6 +121,8 @@ type Snapshot interface {
 	// Storage directly retrieves the storage data associated with a particular hash,
 	// within a particular account.
 	Storage(accountHash, storageHash common.Hash) ([]byte, error)
+
+	Generating() bool
 }
 
 // snapshot is the internal version of the snapshot data layer that supports some