Merge pull request #831 from cffls/block-stm

Run serial and parallel processor at the same time

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)
+	blockImportTimer              = metrics.NewRegisteredMeter("chain/imports", nil)
-	blockInsertTimer     = metrics.NewRegisteredTimer("chain/inserts", nil)
+	blockInsertTimer              = metrics.NewRegisteredTimer("chain/inserts", nil)
-	blockValidationTimer = metrics.NewRegisteredTimer("chain/validation", nil)
+	blockValidationTimer          = metrics.NewRegisteredTimer("chain/validation", nil)
-	blockExecutionTimer  = metrics.NewRegisteredTimer("chain/execution", nil)
+	blockExecutionTimer           = metrics.NewRegisteredTimer("chain/execution", nil)
-	blockWriteTimer      = metrics.NewRegisteredTimer("chain/write", 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
+	engine            consensus.Engine
-	validator  Validator // Block and state validator interface
+	validator         Validator // Block and state validator interface
-	prefetcher Prefetcher
+	prefetcher        Prefetcher
-	processor  Processor // Block transaction processor interface
+	processor         Processor // Block transaction processor interface
-	forker     *ForkChoice
+	parallelProcessor Processor // Parallel block transaction processor interface
-	vmConfig   vm.Config
+	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)

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())
+	if err := testBlockChainImport(blockChainB, blockchain); err == nil {
-	cancel()
+		t.Fatalf("expected error for bad tx")
-
-	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)
 	}
 }
 
@@ -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 {

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 {
+		if interruptCtx != nil && interruptCtx.Err() != nil {
-			select {
+			pe.Close(true)
-			case <-interruptCtx.Done():
+			return result, interruptCtx.Err()
-				pe.Close(true)
-				return result, interruptCtx.Err()
-			default:
-			}
 		}
 
 		res := pe.resultQueue.Pop().(ExecResult)

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")
+	}
+}

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
 	}