Remove dependency when a transaction is reverted

If a transaction is reverted, its write records should be excluded from
dependency calculation.

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

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,

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

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

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))
+		RevertWrite(s, blockstm.NewSubpathKey(*ch.account, SuicidePath))
-		MVWrite(s, blockstm.NewSubpathKey(*ch.account, BalancePath))
+		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 {

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
+	mvHashmap    *blockstm.MVHashMap
-	incarnation     int
+	incarnation  int
-	readMap         map[blockstm.Key]blockstm.ReadDescriptor
+	readMap      map[blockstm.Key]blockstm.ReadDescriptor
-	writeMap        map[blockstm.Key]blockstm.WriteDescriptor
+	writeMap     map[blockstm.Key]blockstm.WriteDescriptor
-	newStateObjects map[common.Address]struct{}
+	revertedKeys map[blockstm.Key]struct{}
-	dep             int
+	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,