diff --git a/core/blockchain.go b/core/blockchain.go index fe8172e41e..47164cc5b9 100644 --- a/core/blockchain.go +++ b/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 diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index f0c05a7d94..0e2e8af137 100644 --- a/core/blockstm/executor.go +++ b/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 { - err = res.err + if abortErr, ok := res.err.(ErrExecAbortError); ok && abortErr.OriginError != nil && pe.skipCheck[tx] { + // 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) - close(pe.chSpeculativeTasks) - pe.workerWg.Wait() - close(pe.chResults) - pe.settleWg.Wait() - close(pe.chSettle) + pe.Close(true) 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 nil - }) + return executeParallelWithCheck(tasks, profile, nil) } diff --git a/core/blockstm/executor_test.go b/core/blockstm/executor_test.go index e7f6d685f3..bed60cdcd4 100644 --- a/core/blockstm/executor_test.go +++ b/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 diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 871c38668f..3a530852fa 100644 --- a/core/parallel_state_processor.go +++ b/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 } diff --git a/core/state_processor_test.go b/core/state_processor_test.go index 5dc076a11c..5eb7938811 100644 --- a/core/state_processor_test.go +++ b/core/state_processor_test.go @@ -234,28 +234,33 @@ func TestStateProcessorErrors(t *testing.T) { }, }, } - genesis = gspec.MustCommit(db) - blockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil) + genesis = gspec.MustCommit(db) + 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 { - txs []*types.Transaction - want string - }{ - { // ErrTxTypeNotSupported - txs: []*types.Transaction{ - mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)), + defer parallelBlockchain.Stop() + + for _, bc := range []*BlockChain{blockchain, parallelBlockchain} { + for i, tt := range []struct { + txs []*types.Transaction + want string + }{ + { // 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 := blockchain.InsertChain(types.Blocks{block}) - 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) + } { + block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config) + _, err := bc.InsertChain(types.Blocks{block}) + 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) + } } } } @@ -274,28 +279,33 @@ func TestStateProcessorErrors(t *testing.T) { }, }, } - genesis = gspec.MustCommit(db) - blockchain, _ = NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil, nil) + genesis = gspec.MustCommit(db) + 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 { - txs []*types.Transaction - want string - }{ - { // ErrSenderNoEOA - txs: []*types.Transaction{ - mkDynamicTx(0, common.Address{}, params.TxGas-1000, big.NewInt(0), big.NewInt(0)), + defer parallelBlockchain.Stop() + + for _, bc := range []*BlockChain{blockchain, parallelBlockchain} { + for i, tt := range []struct { + txs []*types.Transaction + want string + }{ + { // 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 := blockchain.InsertChain(types.Blocks{block}) - 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) + } { + block := GenerateBadBlock(genesis, ethash.NewFaker(), tt.txs, gspec.Config) + _, err := bc.InsertChain(types.Blocks{block}) + 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) + } } } }