diff --git a/core/blockchain.go b/core/blockchain.go index 92d74a8be7..6a44a3acff 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -42,6 +42,7 @@ import ( "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/common/tracing" "github.com/ethereum/go-ethereum/consensus" + "github.com/ethereum/go-ethereum/core/blockstm" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/state/snapshot" @@ -76,11 +77,13 @@ var ( snapshotStorageReadTimer = metrics.NewRegisteredTimer("chain/snapshot/storage/reads", nil) snapshotCommitTimer = metrics.NewRegisteredTimer("chain/snapshot/commits", nil) - blockImportTimer = metrics.NewRegisteredMeter("chain/imports", nil) - blockInsertTimer = metrics.NewRegisteredTimer("chain/inserts", nil) - blockValidationTimer = metrics.NewRegisteredTimer("chain/validation", nil) - blockExecutionTimer = metrics.NewRegisteredTimer("chain/execution", nil) - blockWriteTimer = metrics.NewRegisteredTimer("chain/write", nil) + blockImportTimer = metrics.NewRegisteredMeter("chain/imports", nil) + blockInsertTimer = metrics.NewRegisteredTimer("chain/inserts", nil) + blockValidationTimer = metrics.NewRegisteredTimer("chain/validation", nil) + blockExecutionTimer = metrics.NewRegisteredTimer("chain/execution", nil) + blockWriteTimer = metrics.NewRegisteredTimer("chain/write", nil) + blockExecutionParallelCounter = metrics.NewRegisteredCounter("chain/execution/parallel", nil) + blockExecutionSerialCounter = metrics.NewRegisteredCounter("chain/execution/serial", nil) blockReorgMeter = metrics.NewRegisteredMeter("chain/reorg/executes", nil) blockReorgAddMeter = metrics.NewRegisteredMeter("chain/reorg/add", nil) @@ -216,12 +219,13 @@ type BlockChain struct { running int32 // 0 if chain is running, 1 when stopped procInterrupt int32 // interrupt signaler for block processing - engine consensus.Engine - validator Validator // Block and state validator interface - prefetcher Prefetcher - processor Processor // Block transaction processor interface - forker *ForkChoice - vmConfig vm.Config + engine consensus.Engine + validator Validator // Block and state validator interface + prefetcher Prefetcher + processor Processor // Block transaction processor interface + parallelProcessor Processor // Parallel block transaction processor interface + forker *ForkChoice + vmConfig vm.Config // Bor related changes borReceiptsCache *lru.Cache // Cache for the most recent bor receipt receipts per block @@ -443,11 +447,85 @@ func NewParallelBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainCon return nil, err } - bc.processor = NewParallelStateProcessor(chainConfig, bc, engine) + bc.parallelProcessor = NewParallelStateProcessor(chainConfig, bc, engine) return bc, nil } +func (bc *BlockChain) ProcessBlock(block *types.Block, parent *types.Header) (types.Receipts, []*types.Log, uint64, *state.StateDB, error) { + // Process the block using processor and parallelProcessor at the same time, take the one which finishes first, cancel the other, and return the result + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + type Result struct { + receipts types.Receipts + logs []*types.Log + usedGas uint64 + err error + statedb *state.StateDB + counter metrics.Counter + } + + resultChan := make(chan Result, 2) + + processorCount := 0 + + if bc.parallelProcessor != nil { + parallelStatedb, err := state.New(parent.Root, bc.stateCache, bc.snaps) + if err != nil { + return nil, nil, 0, nil, err + } + + processorCount++ + + go func() { + parallelStatedb.StartPrefetcher("chain") + receipts, logs, usedGas, err := bc.parallelProcessor.Process(block, parallelStatedb, bc.vmConfig, ctx) + resultChan <- Result{receipts, logs, usedGas, err, parallelStatedb, blockExecutionParallelCounter} + }() + } + + if bc.processor != nil { + statedb, err := state.New(parent.Root, bc.stateCache, bc.snaps) + if err != nil { + return nil, nil, 0, nil, err + } + + processorCount++ + + go func() { + statedb.StartPrefetcher("chain") + receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig, ctx) + resultChan <- Result{receipts, logs, usedGas, err, statedb, blockExecutionSerialCounter} + }() + } + + result := <-resultChan + + if _, ok := result.err.(blockstm.ParallelExecFailedError); ok { + log.Warn("Parallel state processor failed", "err", result.err) + + // If the parallel processor failed, we will fallback to the serial processor if enabled + if processorCount == 2 { + result.statedb.StopPrefetcher() + result = <-resultChan + processorCount-- + } + } + + result.counter.Inc(1) + + // Make sure we are not leaking any prefetchers + if processorCount == 2 { + go func() { + second_result := <-resultChan + second_result.statedb.StopPrefetcher() + }() + } + + return result.receipts, result.logs, result.usedGas, result.statedb, result.err +} + // 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 @@ -1774,14 +1852,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals, setHead bool) if parent == nil { parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) } - statedb, err := state.New(parent.Root, bc.stateCache, bc.snaps) - if err != nil { - return it.index, err - } - - // Enable prefetching to pull in trie node paths while processing transactions - statedb.StartPrefetcher("chain") - activeState = statedb // If we have a followup block, run that against the current state to pre-cache // transactions and probabilistically some of the account/storage trie nodes. @@ -1803,7 +1873,8 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals, setHead bool) // Process block using the parent state as reference point substart := time.Now() - receipts, logs, usedGas, err := bc.processor.Process(block, statedb, bc.vmConfig, nil) + receipts, logs, usedGas, statedb, err := bc.ProcessBlock(block, parent) + activeState = statedb if err != nil { bc.reportBlock(block, receipts, err) atomic.StoreUint32(&followupInterrupt, 1) diff --git a/core/blockchain_test.go b/core/blockchain_test.go index 3c80ef3718..88093accab 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -17,7 +17,6 @@ package core import ( - "context" "errors" "fmt" "io/ioutil" @@ -34,7 +33,6 @@ import ( "github.com/ethereum/go-ethereum/consensus/beacon" "github.com/ethereum/go-ethereum/consensus/ethash" "github.com/ethereum/go-ethereum/core/rawdb" - "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/crypto" @@ -125,7 +123,7 @@ func testFork(t *testing.T, blockchain *BlockChain, i, n int, full bool, compara cur := blockchain.CurrentBlock() tdPre = blockchain.GetTd(cur.Hash(), cur.NumberU64()) - if err := testBlockChainImport(blockChainB, blockchain, nil); err != nil { + if err := testBlockChainImport(blockChainB, blockchain); err != nil { t.Fatalf("failed to import forked block chain: %v", err) } @@ -146,7 +144,7 @@ func testFork(t *testing.T, blockchain *BlockChain, i, n int, full bool, compara // testBlockChainImport tries to process a chain of blocks, writing them into // the database if successful. -func testBlockChainImport(chain types.Blocks, blockchain *BlockChain, ctx context.Context) error { +func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error { for _, block := range chain { // Try and process the block err := blockchain.engine.VerifyHeader(blockchain, block.Header(), true) @@ -160,13 +158,8 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain, ctx contex return err } - statedb, err := state.New(blockchain.GetBlockByHash(block.ParentHash()).Root(), blockchain.stateCache, nil) + receipts, _, usedGas, statedb, err := blockchain.ProcessBlock(block, blockchain.GetBlockByHash(block.ParentHash()).Header()) - if err != nil { - return err - } - - receipts, _, usedGas, err := blockchain.processor.Process(block, statedb, vm.Config{}, ctx) if err != nil { blockchain.reportBlock(block, receipts, err) return err @@ -186,10 +179,12 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain, ctx contex return nil } -func TestBlockChainImportInterrupt(t *testing.T) { +func TestParallelBlockChainImport(t *testing.T) { t.Parallel() db, blockchain, err := newCanonical(ethash.NewFaker(), 10, true) + blockchain.parallelProcessor = NewParallelStateProcessor(blockchain.chainConfig, blockchain, blockchain.engine) + if err != nil { t.Fatalf("failed to make new canonical chain: %v", err) } @@ -198,11 +193,8 @@ func TestBlockChainImportInterrupt(t *testing.T) { blockChainB := makeFakeNonEmptyBlockChain(blockchain.CurrentBlock(), 5, ethash.NewFaker(), db, forkSeed, 5) - ctx, cancel := context.WithCancel(context.Background()) - cancel() - - if err := testBlockChainImport(blockChainB, blockchain, ctx); err != context.Canceled { - t.Errorf("block chain import is not cancelled correctly, got %v, want %v", err, context.Canceled) + if err := testBlockChainImport(blockChainB, blockchain); err == nil { + t.Fatalf("expected error for bad tx") } } @@ -502,7 +494,7 @@ func testBrokenChain(t *testing.T, full bool) { // Create a forked chain, and try to insert with a missing link if full { chain := makeBlockChain(blockchain.CurrentBlock(), 5, ethash.NewFaker(), db, forkSeed)[1:] - if err := testBlockChainImport(chain, blockchain, nil); err == nil { + if err := testBlockChainImport(chain, blockchain); err == nil { t.Errorf("broken block chain not reported") } } else { diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index 0e7a06771d..a685fe0707 100644 --- a/core/blockstm/executor.go +++ b/core/blockstm/executor.go @@ -69,6 +69,14 @@ func (e ErrExecAbortError) Error() string { } } +type ParallelExecFailedError struct { + Msg string +} + +func (e ParallelExecFailedError) Error() string { + return e.Msg +} + type IntHeap []int func (h IntHeap) Len() int { return len(h) } @@ -290,7 +298,7 @@ func NewParallelExecutor(tasks []ExecTask, profile bool, metadata bool) *Paralle } // nolint: gocognit -func (pe *ParallelExecutor) Prepare() { +func (pe *ParallelExecutor) Prepare() error { prevSenderTx := make(map[common.Address]int) for i, t := range pe.tasks { @@ -382,11 +390,16 @@ func (pe *ParallelExecutor) Prepare() { // bootstrap first execution tx := pe.execTasks.takeNextPending() - if tx != -1 { - pe.cntExec++ - pe.chTasks <- ExecVersionView{ver: Version{tx, 0}, et: pe.tasks[tx], mvh: pe.mvh, sender: pe.tasks[tx].Sender()} + if tx == -1 { + return ParallelExecFailedError{"no executable transactions due to bad dependency"} } + + pe.cntExec++ + + pe.chTasks <- ExecVersionView{ver: Version{tx, 0}, et: pe.tasks[tx], mvh: pe.mvh, sender: pe.tasks[tx].Sender()} + + return nil } func (pe *ParallelExecutor) Close(wait bool) { @@ -587,16 +600,17 @@ func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyChec } pe := NewParallelExecutor(tasks, profile, metadata) - pe.Prepare() + err = pe.Prepare() + + if err != nil { + pe.Close(true) + return + } for range pe.chResults { - if interruptCtx != nil { - select { - case <-interruptCtx.Done(): - pe.Close(true) - return result, interruptCtx.Err() - default: - } + if interruptCtx != nil && interruptCtx.Err() != nil { + pe.Close(true) + return result, interruptCtx.Err() } res := pe.resultQueue.Pop().(ExecResult) diff --git a/core/blockstm/executor_test.go b/core/blockstm/executor_test.go index fb1f54ed2a..511d9c774a 100644 --- a/core/blockstm/executor_test.go +++ b/core/blockstm/executor_test.go @@ -1,6 +1,7 @@ package blockstm import ( + "context" "fmt" "math/big" "math/rand" @@ -457,7 +458,6 @@ func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck, metad func runParallelGetMetadata(t *testing.T, tasks []ExecTask, validation PropertyCheck) map[int]map[int]bool { t.Helper() - // fmt.Println("len(tasks)", len(tasks)) res, err := executeParallelWithCheck(tasks, true, validation, false, nil) assert.NoError(t, err, "error occur during parallel execution") @@ -923,3 +923,62 @@ func TestDexScenarioWithMetadata(t *testing.T) { testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } + +func TestBreakFromCircularDependency(t *testing.T) { + t.Parallel() + rand.Seed(0) + + tasks := make([]ExecTask, 5) + + for i := range tasks { + tasks[i] = &testExecTask{ + txIdx: i, + dependencies: []int{ + (i + len(tasks) - 1) % len(tasks), + }, + } + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // This should not hang + _, err := ExecuteParallel(tasks, false, true, ctx) + + if err == nil { + t.Error("Expected cancel error") + } +} + +func TestBreakFromPartialCircularDependency(t *testing.T) { + t.Parallel() + rand.Seed(0) + + tasks := make([]ExecTask, 5) + + for i := range tasks { + if i < 3 { + tasks[i] = &testExecTask{ + txIdx: i, + dependencies: []int{ + (i + 2) % 3, + }, + } + } else { + tasks[i] = &testExecTask{ + txIdx: i, + dependencies: []int{}, + } + } + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // This should not hang + _, err := ExecuteParallel(tasks, false, true, ctx) + + if err == nil { + t.Error("Expected cancel error") + } +} diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 805a89646d..c84427ec4f 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -87,8 +87,7 @@ type ExecutionTask struct { // (0 -> delay is not allowed, 1 -> delay is allowed) // next k elements in dependencies -> transaction indexes on which transaction i is dependent on dependencies []int - - coinbase common.Address + coinbase common.Address } func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (err error) { @@ -339,6 +338,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat receipts: &receipts, allLogs: &allLogs, dependencies: deps[i], + coinbase: coinbase, } tasks = append(tasks, task) @@ -394,6 +394,8 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat task := task.(*ExecutionTask) if task.shouldRerunWithoutFeeDelay { shouldDelayFeeCal = false + + statedb.StopPrefetcher() *statedb = *backupStateDB allLogs = []*types.Log{} @@ -415,10 +417,6 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat } if err != nil { - if err != context.Canceled { - log.Error("blockstm error executing block", "err", err) - } - return nil, nil, 0, err }