Merge block-stm to develop

builder/files/config.toml

@@ -171,6 +171,10 @@ syncmode = "full"
   # period = 0
   # gaslimit = 11500000
 
+# [parallelevm]
+  # enable = false
+  # procs = 8
+
 # [pprof]
 #   pprof = false
 #   port = 6060

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
@@ -435,6 +439,93 @@ 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.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
@@ -1761,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.
@@ -1790,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)
+		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

@@ -33,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"
@@ -123,9 +122,11 @@ func testFork(t *testing.T, blockchain *BlockChain, i, n int, full bool, compara
 	if full {
 		cur := blockchain.CurrentBlock()
 		tdPre = blockchain.GetTd(cur.Hash(), cur.NumberU64())
+
 		if err := testBlockChainImport(blockChainB, blockchain); err != nil {
 			t.Fatalf("failed to import forked block chain: %v", err)
 		}
+
 		last := blockChainB[len(blockChainB)-1]
 		tdPost = blockchain.GetTd(last.Hash(), last.NumberU64())
 	} else {
@@ -156,11 +157,9 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error {
 			}
 			return err
 		}
-		statedb, err := state.New(blockchain.GetBlockByHash(block.ParentHash()).Root(), blockchain.stateCache, nil)
-		if err != nil {
-			return err
-		}
-		receipts, _, usedGas, err := blockchain.processor.Process(block, statedb, vm.Config{})
+
+		receipts, _, usedGas, statedb, err := blockchain.ProcessBlock(block, blockchain.GetBlockByHash(block.ParentHash()).Header())
+
 		if err != nil {
 			blockchain.reportBlock(block, receipts, err)
 			return err
@@ -180,6 +179,25 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error {
 	return nil
 }
 
+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)
+	}
+
+	defer blockchain.Stop()
+
+	blockChainB := makeFakeNonEmptyBlockChain(blockchain.CurrentBlock(), 5, ethash.NewFaker(), db, forkSeed, 5)
+
+	if err := testBlockChainImport(blockChainB, blockchain); err == nil {
+		t.Fatalf("expected error for bad tx")
+	}
+}
+
 // testHeaderChainImport tries to process a chain of header, writing them into
 // the database if successful.
 func testHeaderChainImport(chain []*types.Header, blockchain *BlockChain) error {