[WIP] the Implementaion of Parallel EVM 2.0

cmd/geth/main.go

@@ -165,7 +165,6 @@ var (
 		utils.EVMInterpreterFlag,
 		utils.ParallelTxFlag,
 		utils.ParallelTxNumFlag,
-		utils.ParallelTxQueueSizeFlag,
 		utils.MinerNotifyFullFlag,
 		configFileFlag,
 		utils.CatalystFlag,

cmd/utils/flags.go

@@ -811,11 +811,6 @@ var (
 		Name:  "parallel.num",
 		Usage: "Number of slot for transaction execution, only valid in parallel mode (runtime calculated, no fixed default value)",
 	}
-	ParallelTxQueueSizeFlag = cli.IntFlag{
-		Name:  "parallel.queuesize",
-		Usage: "Max number of Tx that can be queued to a slot, only valid in parallel mode (advanced option)",
-		Value: 20,
-	}
 
 	// Init network
 	InitNetworkSize = cli.IntFlag{
@@ -1675,12 +1670,6 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 			parallelNum = 8 // we found concurrency 8 is slightly better than 15
 		}
 		cfg.ParallelTxNum = parallelNum
-		// set up queue size, it is an advanced option
-		if ctx.GlobalIsSet(ParallelTxQueueSizeFlag.Name) {
-			cfg.ParallelTxQueueSize = ctx.GlobalInt(ParallelTxQueueSizeFlag.Name)
-		} else {
-			cfg.ParallelTxQueueSize = 20 // default queue size, will be optimized
-		}
 	}
 	// Read the value from the flag no matter if it's set or not.
 	cfg.Preimages = ctx.GlobalBool(CachePreimagesFlag.Name)

core/blockchain.go

@@ -3110,15 +3110,15 @@ func EnablePersistDiff(limit uint64) BlockChainOption {
 	}
 }
 
-func EnableParallelProcessor(parallelNum int, queueSize int) BlockChainOption {
+func EnableParallelProcessor(parallelNum int) BlockChainOption {
 	return func(chain *BlockChain) *BlockChain {
 		if chain.snaps == nil {
 			// disable parallel processor if snapshot is not enabled to avoid concurrent issue for SecureTrie
 			log.Info("parallel processor is not enabled since snapshot is not enabled")
 			return chain
 		}
 		chain.parallelExecution = true
-		chain.processor = NewParallelStateProcessor(chain.Config(), chain, chain.engine, parallelNum, queueSize)
+		chain.processor = NewParallelStateProcessor(chain.Config(), chain, chain.engine, parallelNum)
 		return chain
 	}
 }

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,19 +160,28 @@ 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 {
+		s.snapParallelLock.Lock()
 		delete(s.snapDestructs, ch.prev.address)
+		s.snapParallelLock.Unlock()
 	}
 }
 
 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 +194,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 +258,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 +287,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)
 	}