diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index a685fe0707..7ce99f9492 100644 --- a/core/blockstm/executor.go +++ b/core/blockstm/executor.go @@ -556,11 +556,14 @@ func (pe *ParallelExecutor) Step(res *ExecResult) (result ParallelExecutionResul var allDeps map[int]map[int]bool + var deps DAG + if pe.profile { allDeps = GetDep(*pe.lastTxIO) + deps = BuildDAG(*pe.lastTxIO) } - return ParallelExecutionResult{pe.lastTxIO, &pe.stats, nil, allDeps}, err + return ParallelExecutionResult{pe.lastTxIO, &pe.stats, &deps, allDeps}, err } // Send the next immediate pending transaction to be executed diff --git a/core/blockstm/mvhashmap.go b/core/blockstm/mvhashmap.go index 2a517bcc84..fdaece261f 100644 --- a/core/blockstm/mvhashmap.go +++ b/core/blockstm/mvhashmap.go @@ -269,6 +269,13 @@ func ValidateVersion(txIdx int, lastInputOutput *TxnInputOutput, versionedData * mvResult := versionedData.Read(rd.Path, txIdx) switch mvResult.Status() { case MVReadResultDone: + // Having a write record for a path in MVHashmap doesn't necessarily mean there is a conflict, because MVHashmap + // is a superset of the actual write set. + // Check if the write record is actually in write set. If not, skip the key. + if mvResult.depIdx >= 0 && !lastInputOutput.HasWritten(mvResult.depIdx, rd.Path) { + continue + } + valid = rd.Kind == ReadKindMap && rd.V == Version{ TxnIndex: mvResult.depIdx, Incarnation: mvResult.incarnation, diff --git a/core/blockstm/txio.go b/core/blockstm/txio.go index 22541e0a78..19955fb152 100644 --- a/core/blockstm/txio.go +++ b/core/blockstm/txio.go @@ -46,6 +46,7 @@ func (txo TxnOutput) hasNewWrite(cmpSet []WriteDescriptor) bool { type TxnInputOutput struct { inputs []TxnInput outputs []TxnOutput // write sets that should be checked during validation + outputsSet []map[Key]struct{} allOutputs []TxnOutput // entire write sets in MVHashMap. allOutputs should always be a parent set of outputs } @@ -61,10 +62,16 @@ func (io *TxnInputOutput) AllWriteSet(txnIdx int) []WriteDescriptor { return io.allOutputs[txnIdx] } +func (io *TxnInputOutput) HasWritten(txnIdx int, k Key) bool { + _, ok := io.outputsSet[txnIdx][k] + return ok +} + func MakeTxnInputOutput(numTx int) *TxnInputOutput { return &TxnInputOutput{ inputs: make([]TxnInput, numTx), outputs: make([]TxnOutput, numTx), + outputsSet: make([]map[Key]struct{}, numTx), allOutputs: make([]TxnOutput, numTx), } } @@ -75,6 +82,11 @@ func (io *TxnInputOutput) recordRead(txId int, input []ReadDescriptor) { func (io *TxnInputOutput) recordWrite(txId int, output []WriteDescriptor) { io.outputs[txId] = output + io.outputsSet[txId] = make(map[Key]struct{}, len(output)) + + for _, v := range output { + io.outputsSet[txId][v.Path] = struct{}{} + } } func (io *TxnInputOutput) recordAllWrite(txId int, output []WriteDescriptor) { diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index e72fb4e6d8..8956f6f4d2 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -178,7 +178,7 @@ func (task *ExecutionTask) Settle() { coinbaseBalance := task.finalStateDB.GetBalance(task.coinbase) - task.finalStateDB.ApplyMVWriteSet(task.statedb.MVWriteList()) + task.finalStateDB.ApplyMVWriteSet(task.statedb.MVFullWriteList()) for _, l := range task.statedb.GetLogs(task.tx.Hash(), task.blockHash) { task.finalStateDB.AddLog(l) @@ -290,11 +290,6 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat if block.Header().TxDependency != nil { metadata = true - - if !VerifyDeps(deps) { - metadata = false - deps = GetDeps([][]uint64{}) - } } blockContext := NewEVMBlockContext(header, p.bc, nil) @@ -369,9 +364,9 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat backupStateDB := statedb.Copy() profile := false - result, err := blockstm.ExecuteParallel(tasks, false, metadata, interruptCtx) + result, err := blockstm.ExecuteParallel(tasks, profile, metadata, interruptCtx) - if err == nil && profile { + if err == nil && profile && result.Deps != nil { _, weight := result.Deps.LongestPath(*result.Stats) serialWeight := uint64(0) @@ -381,10 +376,6 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat } parallelizabilityTimer.Update(time.Duration(serialWeight * 100 / weight)) - - log.Info("Parallelizability", "Average (%)", parallelizabilityTimer.Mean()) - - log.Info("Parallelizability", "Histogram (%)", parallelizabilityTimer.Percentiles([]float64{0.001, 0.01, 0.05, 0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 0.99, 0.999, 0.9999})) } for _, task := range tasks { @@ -436,20 +427,3 @@ func GetDeps(txDependency [][]uint64) map[int][]int { return deps } - -// returns true if dependencies are correct -func VerifyDeps(deps map[int][]int) bool { - // number of transactions in the block - n := len(deps) - - // Handle out-of-range and circular dependency problem - for tx, val := range deps { - for depTx := range val { - if depTx >= n || depTx < tx { - return false - } - } - } - - return true -} diff --git a/core/state/journal.go b/core/state/journal.go index 79a4e35422..2a0c81dcb6 100644 --- a/core/state/journal.go +++ b/core/state/journal.go @@ -144,7 +144,7 @@ type ( func (ch createObjectChange) revert(s *StateDB) { delete(s.stateObjects, *ch.account) delete(s.stateObjectsDirty, *ch.account) - MVWrite(s, blockstm.NewAddressKey(*ch.account)) + RevertWrite(s, blockstm.NewAddressKey(*ch.account)) } func (ch createObjectChange) dirtied() *common.Address { @@ -153,7 +153,7 @@ func (ch createObjectChange) dirtied() *common.Address { func (ch resetObjectChange) revert(s *StateDB) { s.setStateObject(ch.prev) - MVWrite(s, blockstm.NewAddressKey(ch.prev.address)) + RevertWrite(s, blockstm.NewAddressKey(ch.prev.address)) if !ch.prevdestruct && s.snap != nil { delete(s.snapDestructs, ch.prev.addrHash) } @@ -168,8 +168,8 @@ func (ch suicideChange) revert(s *StateDB) { if obj != nil { obj.suicided = ch.prev obj.setBalance(ch.prevbalance) - MVWrite(s, blockstm.NewSubpathKey(*ch.account, SuicidePath)) - MVWrite(s, blockstm.NewSubpathKey(*ch.account, BalancePath)) + RevertWrite(s, blockstm.NewSubpathKey(*ch.account, SuicidePath)) + RevertWrite(s, blockstm.NewSubpathKey(*ch.account, BalancePath)) } } @@ -188,7 +188,7 @@ func (ch touchChange) dirtied() *common.Address { func (ch balanceChange) revert(s *StateDB) { s.getStateObject(*ch.account).setBalance(ch.prev) - MVWrite(s, blockstm.NewSubpathKey(*ch.account, BalancePath)) + RevertWrite(s, blockstm.NewSubpathKey(*ch.account, BalancePath)) } func (ch balanceChange) dirtied() *common.Address { @@ -197,7 +197,7 @@ func (ch balanceChange) dirtied() *common.Address { func (ch nonceChange) revert(s *StateDB) { s.getStateObject(*ch.account).setNonce(ch.prev) - MVWrite(s, blockstm.NewSubpathKey(*ch.account, NoncePath)) + RevertWrite(s, blockstm.NewSubpathKey(*ch.account, NoncePath)) } func (ch nonceChange) dirtied() *common.Address { @@ -206,7 +206,7 @@ func (ch nonceChange) dirtied() *common.Address { func (ch codeChange) revert(s *StateDB) { s.getStateObject(*ch.account).setCode(common.BytesToHash(ch.prevhash), ch.prevcode) - MVWrite(s, blockstm.NewSubpathKey(*ch.account, CodePath)) + RevertWrite(s, blockstm.NewSubpathKey(*ch.account, CodePath)) } func (ch codeChange) dirtied() *common.Address { @@ -215,7 +215,7 @@ func (ch codeChange) dirtied() *common.Address { func (ch storageChange) revert(s *StateDB) { s.getStateObject(*ch.account).setState(ch.key, ch.prevalue) - MVWrite(s, blockstm.NewStateKey(*ch.account, ch.key)) + RevertWrite(s, blockstm.NewStateKey(*ch.account, ch.key)) } func (ch storageChange) dirtied() *common.Address { diff --git a/core/state/statedb.go b/core/state/statedb.go index b77bd9d763..881ea8c110 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -81,12 +81,12 @@ type StateDB struct { stateObjectsDirty map[common.Address]struct{} // State objects modified in the current execution // Block-stm related fields - mvHashmap *blockstm.MVHashMap - incarnation int - readMap map[blockstm.Key]blockstm.ReadDescriptor - writeMap map[blockstm.Key]blockstm.WriteDescriptor - newStateObjects map[common.Address]struct{} - dep int + mvHashmap *blockstm.MVHashMap + incarnation int + readMap map[blockstm.Key]blockstm.ReadDescriptor + writeMap map[blockstm.Key]blockstm.WriteDescriptor + revertedKeys map[blockstm.Key]struct{} + dep int // DB error. // State objects are used by the consensus core and VM which are @@ -147,7 +147,7 @@ func New(root common.Hash, db Database, snaps *snapshot.Tree) (*StateDB, error) stateObjects: make(map[common.Address]*stateObject), stateObjectsPending: make(map[common.Address]struct{}), stateObjectsDirty: make(map[common.Address]struct{}), - newStateObjects: make(map[common.Address]struct{}), + revertedKeys: make(map[blockstm.Key]struct{}), logs: make(map[common.Hash][]*types.Log), preimages: make(map[common.Hash][]byte), journal: newJournal(), @@ -187,7 +187,7 @@ func (s *StateDB) MVWriteList() []blockstm.WriteDescriptor { writes := make([]blockstm.WriteDescriptor, 0, len(s.writeMap)) for _, v := range s.writeMap { - if !v.Path.IsAddress() { + if _, ok := s.revertedKeys[v.Path]; !ok { writes = append(writes, v) } } @@ -212,7 +212,7 @@ func (s *StateDB) MVReadMap() map[blockstm.Key]blockstm.ReadDescriptor { func (s *StateDB) MVReadList() []blockstm.ReadDescriptor { reads := make([]blockstm.ReadDescriptor, 0, len(s.readMap)) - for _, v := range s.readMap { + for _, v := range s.MVReadMap() { reads = append(reads, v) } @@ -268,6 +268,14 @@ func MVRead[T any](s *StateDB, k blockstm.Key, defaultV T, readStorage func(s *S } } + if !k.IsAddress() { + // If we are reading subpath from a deleted account, return default value instead of reading from MVHashmap + addr := k.GetAddress() + if s.getStateObject(addr) == nil { + return defaultV + } + } + res := s.mvHashmap.Read(k, s.txIndex) var rd blockstm.ReadDescriptor @@ -320,6 +328,10 @@ func MVWrite(s *StateDB, k blockstm.Key) { } } +func RevertWrite(s *StateDB, k blockstm.Key) { + s.revertedKeys[k] = struct{}{} +} + func MVWritten(s *StateDB, k blockstm.Key) bool { if s.mvHashmap == nil || s.writeMap == nil { return false @@ -349,6 +361,8 @@ func (sw *StateDB) ApplyMVWriteSet(writes []blockstm.WriteDescriptor) { stateKey := path.GetStateKey() state := sr.GetState(addr, stateKey) sw.SetState(addr, stateKey, state) + } else if path.IsAddress() { + continue } else { addr := path.GetAddress() switch path.GetSubpath() { @@ -543,10 +557,6 @@ const SuicidePath = 4 // GetBalance retrieves the balance from the given address or 0 if object not found func (s *StateDB) GetBalance(addr common.Address) *big.Int { - if s.getStateObject(addr) == nil { - return common.Big0 - } - return MVRead(s, blockstm.NewSubpathKey(addr, BalancePath), common.Big0, func(s *StateDB) *big.Int { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -558,10 +568,6 @@ func (s *StateDB) GetBalance(addr common.Address) *big.Int { } func (s *StateDB) GetNonce(addr common.Address) uint64 { - if s.getStateObject(addr) == nil { - return 0 - } - return MVRead(s, blockstm.NewSubpathKey(addr, NoncePath), 0, func(s *StateDB) uint64 { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -585,10 +591,6 @@ func (s *StateDB) Version() blockstm.Version { } func (s *StateDB) GetCode(addr common.Address) []byte { - if s.getStateObject(addr) == nil { - return nil - } - return MVRead(s, blockstm.NewSubpathKey(addr, CodePath), nil, func(s *StateDB) []byte { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -599,10 +601,6 @@ func (s *StateDB) GetCode(addr common.Address) []byte { } func (s *StateDB) GetCodeSize(addr common.Address) int { - if s.getStateObject(addr) == nil { - return 0 - } - return MVRead(s, blockstm.NewSubpathKey(addr, CodePath), 0, func(s *StateDB) int { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -613,10 +611,6 @@ func (s *StateDB) GetCodeSize(addr common.Address) int { } func (s *StateDB) GetCodeHash(addr common.Address) common.Hash { - if s.getStateObject(addr) == nil { - return common.Hash{} - } - return MVRead(s, blockstm.NewSubpathKey(addr, CodePath), common.Hash{}, func(s *StateDB) common.Hash { stateObject := s.getStateObject(addr) if stateObject == nil { @@ -628,10 +622,6 @@ func (s *StateDB) GetCodeHash(addr common.Address) common.Hash { // GetState retrieves a value from the given account's storage trie. func (s *StateDB) GetState(addr common.Address, hash common.Hash) common.Hash { - if s.getStateObject(addr) == nil { - return common.Hash{} - } - return MVRead(s, blockstm.NewStateKey(addr, hash), common.Hash{}, func(s *StateDB) common.Hash { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -666,10 +656,6 @@ func (s *StateDB) GetStorageProof(a common.Address, key common.Hash) ([][]byte, // GetCommittedState retrieves a value from the given account's committed storage trie. func (s *StateDB) GetCommittedState(addr common.Address, hash common.Hash) common.Hash { - if s.getStateObject(addr) == nil { - return common.Hash{} - } - return MVRead(s, blockstm.NewStateKey(addr, hash), common.Hash{}, func(s *StateDB) common.Hash { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -697,10 +683,6 @@ func (s *StateDB) StorageTrie(addr common.Address) Trie { } func (s *StateDB) HasSuicided(addr common.Address) bool { - if s.getStateObject(addr) == nil { - return false - } - return MVRead(s, blockstm.NewSubpathKey(addr, SuicidePath), false, func(s *StateDB) bool { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -982,7 +964,6 @@ func (s *StateDB) createObject(addr common.Address) (newobj, prev *stateObject) s.journal.append(resetObjectChange{prev: prev, prevdestruct: prevdestruct}) } s.setStateObject(newobj) - s.newStateObjects[addr] = struct{}{} MVWrite(s, blockstm.NewAddressKey(addr)) if prev != nil && !prev.deleted { @@ -1048,7 +1029,7 @@ func (s *StateDB) Copy() *StateDB { stateObjects: make(map[common.Address]*stateObject, len(s.journal.dirties)), stateObjectsPending: make(map[common.Address]struct{}, len(s.stateObjectsPending)), stateObjectsDirty: make(map[common.Address]struct{}, len(s.journal.dirties)), - newStateObjects: make(map[common.Address]struct{}, len(s.newStateObjects)), + revertedKeys: make(map[blockstm.Key]struct{}), refund: s.refund, logs: make(map[common.Hash][]*types.Log, len(s.logs)), logSize: s.logSize,