Recognize bad transactions and break loop in blockstm executor

core/blockchain.go

@@ -223,6 +223,7 @@ type BlockChain struct {
 // NewBlockChain returns a fully initialised block chain using information
 // available in the database. It initialises the default Ethereum Validator
 // and Processor.
+//
 //nolint:gocognit
 func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(header *types.Header) bool, txLookupLimit *uint64, checker ethereum.ChainValidator) (*BlockChain, error) {
 	if cacheConfig == nil {
@@ -420,6 +421,19 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	return bc, nil
 }
 
+// Similar to NewBlockChain, this function creates a new blockchain object, but with a parallel state processor
+func NewParallelBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(header *types.Header) bool, txLookupLimit *uint64, checker ethereum.ChainValidator) (*BlockChain, error) {
+	bc, err := NewBlockChain(db, cacheConfig, chainConfig, engine, vmConfig, shouldPreserve, txLookupLimit, checker)
+
+	if err != nil {
+		return nil, err
+	}
+
+	bc.processor = NewParallelStateProcessor(chainConfig, bc, engine)
+
+	return bc, nil
+}
+
 // empty returns an indicator whether the blockchain is empty.
 // Note, it's a special case that we connect a non-empty ancient
 // database with an empty node, so that we can plugin the ancient

core/blockstm/executor.go

@@ -23,6 +23,7 @@ type ExecTask interface {
 	MVReadList() []ReadDescriptor
 	MVWriteList() []WriteDescriptor
 	MVFullWriteList() []WriteDescriptor
+	Hash() common.Hash
 	Sender() common.Address
 	Settle()
 }
@@ -48,7 +49,8 @@ func (ev *ExecVersionView) Execute() (er ExecResult) {
 }
 
 type ErrExecAbortError struct {
-	Dependency int
+	Dependency  int
+	OriginError error
 }
 
 func (e ErrExecAbortError) Error() string {
@@ -308,12 +310,33 @@ func (pe *ParallelExecutor) Prepare() {
 	}
 }
 
+func (pe *ParallelExecutor) Close(wait bool) {
+	close(pe.chTasks)
+	close(pe.chSpeculativeTasks)
+
+	if wait {
+		pe.workerWg.Wait()
+	}
+
+	close(pe.chResults)
+
+	if wait {
+		pe.settleWg.Wait()
+	}
+
+	close(pe.chSettle)
+}
+
 // nolint: gocognit
-func (pe *ParallelExecutor) Step(res ExecResult) (result ParallelExecutionResult, err error) {
+func (pe *ParallelExecutor) Step(res *ExecResult) (result ParallelExecutionResult, err error) {
 	tx := res.ver.TxnIndex
 
-	if _, ok := res.err.(ErrExecAbortError); res.err != nil && !ok {
+	if abortErr, ok := res.err.(ErrExecAbortError); ok && abortErr.OriginError != nil && pe.skipCheck[tx] {
-		err = res.err
+		// If the transaction failed when we know it should not fail, this means the transaction itself is
+		// bad (e.g. wrong nonce), and we should exit the execution immediately
+		err = fmt.Errorf("could not apply tx %d [%v]: %w", tx, pe.tasks[tx].Hash(), abortErr.OriginError)
+		pe.Close(false)
+
 		return
 	}
 
@@ -440,12 +463,7 @@ func (pe *ParallelExecutor) Step(res ExecResult) (result ParallelExecutionResult
 	if pe.validateTasks.countComplete() == len(pe.tasks) && pe.execTasks.countComplete() == len(pe.tasks) {
 		log.Debug("blockstm exec summary", "execs", pe.cntExec, "success", pe.cntSuccess, "aborts", pe.cntAbort, "validations", pe.cntTotalValidations, "failures", pe.cntValidationFail, "#tasks/#execs", fmt.Sprintf("%.2f%%", float64(len(pe.tasks))/float64(pe.cntExec)*100))
 
-		close(pe.chTasks)
+		pe.Close(true)
-		close(pe.chSpeculativeTasks)
-		pe.workerWg.Wait()
-		close(pe.chResults)
-		pe.settleWg.Wait()
-		close(pe.chSettle)
 
 		var dag DAG
 
@@ -469,13 +487,9 @@ func (pe *ParallelExecutor) Step(res ExecResult) (result ParallelExecutionResult
 	}
 
 	// Send speculative tasks
-	for pe.execTasks.minPending() != -1 || len(pe.execTasks.inProgress) == 0 {
+	for pe.execTasks.minPending() != -1 {
 		nextTx := pe.execTasks.takeNextPending()
 
-		if nextTx == -1 {
-			nextTx = pe.execTasks.takeNextPending()
-		}
-
 		if nextTx != -1 {
 			pe.cntExec++
 
@@ -502,7 +516,7 @@ func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyChec
 	for range pe.chResults {
 		res := pe.resultQueue.Pop().(ExecResult)
 
-		result, err = pe.Step(res)
+		result, err = pe.Step(&res)
 
 		if err != nil {
 			return result, err
@@ -521,7 +535,5 @@ func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyChec
 }
 
 func ExecuteParallel(tasks []ExecTask, profile bool) (result ParallelExecutionResult, err error) {
-	return executeParallelWithCheck(tasks, profile, func(pe *ParallelExecutor) error {
+	return executeParallelWithCheck(tasks, profile, nil)
-		return nil
-	})
 }

core/blockstm/executor_test.go

@@ -117,7 +117,7 @@ func (t *testExecTask) Execute(mvh *MVHashMap, incarnation int) error {
 	}
 
 	if deps != -1 {
-		return ErrExecAbortError{deps}
+		return ErrExecAbortError{deps, fmt.Errorf("Dependency error")}
 	}
 
 	return nil
@@ -153,6 +153,10 @@ func (t *testExecTask) Sender() common.Address {
 	return t.sender
 }
 
+func (t *testExecTask) Hash() common.Hash {
+	return common.BytesToHash([]byte(fmt.Sprintf("%d", t.txIdx)))
+}
+
 func randTimeGenerator(min time.Duration, max time.Duration) func(txIdx int, opIdx int) time.Duration {
 	return func(txIdx int, opIdx int) time.Duration {
 		return time.Duration(rand.Int63n(int64(max-min))) + min

core/parallel_state_processor.go

@@ -19,7 +19,6 @@ package core
 import (
 	"fmt"
 	"math/big"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus"
@@ -106,7 +105,7 @@ func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (er
 		task.result, err = ApplyMessageNoFeeBurnOrTip(evm, task.msg, new(GasPool).AddGas(task.gasLimit))
 
 		if task.result == nil || err != nil {
-			return blockstm.ErrExecAbortError{Dependency: task.statedb.DepTxIndex()}
+			return blockstm.ErrExecAbortError{Dependency: task.statedb.DepTxIndex(), OriginError: err}
 		}
 
 		reads := task.statedb.MVReadMap()
@@ -127,7 +126,7 @@ func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (er
 	}
 
 	if task.statedb.HadInvalidRead() || err != nil {
-		err = blockstm.ErrExecAbortError{Dependency: task.statedb.DepTxIndex()}
+		err = blockstm.ErrExecAbortError{Dependency: task.statedb.DepTxIndex(), OriginError: err}
 		return
 	}
 
@@ -152,6 +151,10 @@ func (task *ExecutionTask) Sender() common.Address {
 	return task.sender
 }
 
+func (task *ExecutionTask) Hash() common.Hash {
+	return task.tx.Hash()
+}
+
 func (task *ExecutionTask) Settle() {
 	defer func() {
 		if r := recover(); r != nil {
@@ -342,14 +345,8 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat
 		return nil, nil, 0, err
 	}
 
-	statedb.Finalise(p.config.IsEIP158(blockNumber))
-
-	start := time.Now()
-
 	// Finalize the block, applying any consensus engine specific extras (e.g. block rewards)
 	p.engine.Finalize(p.bc, header, statedb, block.Transactions(), block.Uncles())
 
-	fmt.Println("Finalize time of parallel execution:", time.Since(start))
-
 	return receipts, allLogs, *usedGas, nil
 }

core/state_processor_test.go

@@ -234,28 +234,33 @@ func TestStateProcessorErrors(t *testing.T) {
 					},
 				},
 			}
-			genesis       = gspec.MustCommit(db)
+			genesis               = gspec.MustCommit(db)
-			blockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
+			blockchain, _         = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
+			parallelBlockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 		)
 		defer blockchain.Stop()
-		for i, tt := range []struct {
+		defer parallelBlockchain.Stop()
-			txs  []*types.Transaction
+
-			want string
+		for _, bc := range []*BlockChain{blockchain, parallelBlockchain} {
-		}{
+			for i, tt := range []struct {
-			{ // ErrTxTypeNotSupported
+				txs  []*types.Transaction
-				txs: []*types.Transaction{
+				want string
-					mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)),
+			}{
+				{ // ErrTxTypeNotSupported
+					txs: []*types.Transaction{
+						mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)),
+					},
+					want: "could not apply tx 0 [0x88626ac0d53cb65308f2416103c62bb1f18b805573d4f96a3640bbbfff13c14f]: transaction type not supported",
 				},
-				want: "could not apply tx 0 [0x88626ac0d53cb65308f2416103c62bb1f18b805573d4f96a3640bbbfff13c14f]: transaction type not supported",
+			} {
-			},
+				block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config)
-		} {
+				_, err := bc.InsertChain(types.Blocks{block})
-			block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config)
+				if err == nil {
-			_, err := blockchain.InsertChain(types.Blocks{block})
+					t.Fatal("block imported without errors")
-			if err == nil {
+				}
-				t.Fatal("block imported without errors")
+				if have, want := err.Error(), tt.want; have != want {
-			}
+					t.Errorf("test %d:\nhave \"%v\"\nwant \"%v\"\n", i, have, want)
-			if have, want := err.Error(), tt.want; have != want {
+				}
-				t.Errorf("test %d:\nhave \"%v\"\nwant \"%v\"\n", i, have, want)
 			}
 		}
 	}
@@ -274,28 +279,33 @@ func TestStateProcessorErrors(t *testing.T) {
 					},
 				},
 			}
-			genesis       = gspec.MustCommit(db)
+			genesis               = gspec.MustCommit(db)
-			blockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
+			blockchain, _         = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
+			parallelBlockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 		)
 		defer blockchain.Stop()
-		for i, tt := range []struct {
+		defer parallelBlockchain.Stop()
-			txs  []*types.Transaction
+
-			want string
+		for _, bc := range []*BlockChain{blockchain, parallelBlockchain} {
-		}{
+			for i, tt := range []struct {
-			{ // ErrSenderNoEOA
+				txs  []*types.Transaction
-				txs: []*types.Transaction{
+				want string
-					mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)),
+			}{
+				{ // ErrSenderNoEOA
+					txs: []*types.Transaction{
+						mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)),
+					},
+					want: "could not apply tx 0 [0x88626ac0d53cb65308f2416103c62bb1f18b805573d4f96a3640bbbfff13c14f]: sender not an eoa: address 0x71562b71999873DB5b286dF957af199Ec94617F7, codehash: 0x9280914443471259d4570a8661015ae4a5b80186dbc619658fb494bebc3da3d1",
 				},
-				want: "could not apply tx 0 [0x88626ac0d53cb65308f2416103c62bb1f18b805573d4f96a3640bbbfff13c14f]: sender not an eoa: address 0x71562b71999873DB5b286dF957af199Ec94617F7, codehash: 0x9280914443471259d4570a8661015ae4a5b80186dbc619658fb494bebc3da3d1",
+			} {
-			},
+				block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config)
-		} {
+				_, err := bc.InsertChain(types.Blocks{block})
-			block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config)
+				if err == nil {
-			_, err := blockchain.InsertChain(types.Blocks{block})
+					t.Fatal("block imported without errors")
-			if err == nil {
+				}
-				t.Fatal("block imported without errors")
+				if have, want := err.Error(), tt.want; have != want {
-			}
+					t.Errorf("test %d:\nhave \"%v\"\nwant \"%v\"\n", i, have, want)
-			if have, want := err.Error(), tt.want; have != want {
+				}
-				t.Errorf("test %d:\nhave \"%v\"\nwant \"%v\"\n", i, have, want)
 			}
 		}
 	}