pevm: fix some bad check & support to fallback to serial processor;

core/blockchain.go

@@ -1948,15 +1948,16 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 			activeState = statedb
 
 			if bc.vmConfig.EnableParallelExec {
+				bc.parseTxDAG(block)
 				txsCount := block.Transactions().Len()
 				threshold := min(bc.vmConfig.ParallelTxNum/2+2, 4)
-				if txsCount >= threshold {
+				if txsCount < threshold || bc.isEmptyTxDAG() {
+					bc.UseSerialProcessor()
+					log.Debug("Disable Parallel Tx execution", "block", block.NumberU64(), "transactions", txsCount, "parallelTxNum", bc.vmConfig.ParallelTxNum)
+				} else {
 					bc.UseParallelProcessor()
 					statedb.CreateParallelDBManager(2 * txsCount)
 					log.Debug("Enable Parallel Tx execution", "block", block.NumberU64(), "transactions", txsCount, "parallelTxNum", bc.vmConfig.ParallelTxNum)
-				} else {
-					bc.UseSerialProcessor()
-					log.Debug("Disable Parallel Tx execution", "block", block.NumberU64(), "transactions", txsCount, "parallelTxNum", bc.vmConfig.ParallelTxNum)
 				}
 			}
 			// If we have a followup block, run that against the current state to pre-cache
@@ -1982,6 +1983,10 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 			// Process block using the parent state as reference point
 			pstart = time.Now()
 			receipts, logs, usedGas, err = bc.processor.Process(block, statedb, bc.vmConfig)
+			if err == FallbackToSerialProcessorErr {
+				bc.UseSerialProcessor()
+				receipts, logs, usedGas, err = bc.processor.Process(block, statedb, bc.vmConfig)
+			}
 			if err != nil {
 				bc.reportBlock(block, receipts, err)
 				followupInterrupt.Store(true)
@@ -2132,6 +2137,39 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 	return it.index, err
 }
 
+func (bc *BlockChain) parseTxDAG(block *types.Block) {
+	if !bc.enableTxDAG {
+		return
+	}
+	var (
+		txDAG types.TxDAG
+		err   error
+	)
+	if bc.txDAGReader != nil {
+		// load cache txDAG from file first
+		txDAG = bc.txDAGReader.TxDAG(block.NumberU64())
+	} else {
+		// load TxDAG from block
+		txDAG, err = types.GetTxDAG(block)
+		if err != nil {
+			log.Warn("pevm decode txdag failed", "block", block.NumberU64(), "err", err)
+		}
+	}
+	if err := types.ValidateTxDAG(txDAG, len(block.Transactions())); err != nil {
+		log.Warn("pevm cannot apply wrong txdag",
+			"block", block.NumberU64(), "txs", len(block.Transactions()), "err", err)
+		txDAG = nil
+	}
+	bc.vmConfig.TxDAG = txDAG
+}
+
+func (bc *BlockChain) isEmptyTxDAG() bool {
+	if bc.vmConfig.TxDAG != nil && bc.vmConfig.TxDAG.Type() == types.EmptyTxDAGType {
+		return true
+	}
+	return false
+}
+
 // insertSideChain is called when an import batch hits upon a pruned ancestor
 // error, which happens when a sidechain with a sufficiently old fork-block is
 // found.

core/parallel_state_processor.go

@@ -26,6 +26,10 @@ const (
 	stage2AheadNum      = 3  // enter ConfirmStage2 in advance to avoid waiting for Fat Tx
 )
 
+var (
+	FallbackToSerialProcessorErr = errors.New("fallback to serial processor")
+)
+
 type ParallelStateProcessor struct {
 	StateProcessor
 	parallelNum           int          // leave a CPU to dispatcher
@@ -571,11 +575,6 @@ func (p *ParallelStateProcessor) runQuickMergeSlotLoop(slotIndex int, slotType i
 			if txReq.txIndex <= int(p.mergedTxIndex.Load()) {
 				continue
 			}
-
-			if txReq.txIndex != next {
-				log.Warn("query next txReq wrong", "slot", slotIndex, "next", next, "actual", txReq.txIndex)
-				break
-			}
 			if !atomic.CompareAndSwapInt32(&txReq.runnable, 1, 0) {
 				continue
 			}
@@ -773,29 +772,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat
 		ProcessBeaconBlockRoot(*beaconRoot, vmenv, statedb)
 	}
 	statedb.MarkFullProcessed()
-
-	var (
-		txDAG types.TxDAG
-	)
-	if p.bc.enableTxDAG {
-		var err error
-		if p.bc.txDAGReader != nil {
-			// load cache txDAG from file first
-			txDAG = p.bc.txDAGReader.TxDAG(block.NumberU64())
-
-		} else {
-			// load TxDAG from block
-			txDAG, err = types.GetTxDAG(block)
-			if err != nil {
-				log.Debug("pevm decode txdag failed", "block", block.NumberU64(), "err", err)
-			}
-		}
-		if err := types.ValidateTxDAG(txDAG, len(block.Transactions())); err != nil {
-			log.Warn("pevm cannot apply wrong txdag",
-				"block", block.NumberU64(), "txs", len(block.Transactions()), "err", err)
-			txDAG = nil
-		}
-	}
+	txDAG := cfg.TxDAG
 
 	txNum := len(allTxs)
 	latestExcludedTx := -1

core/vm/interpreter.go

@@ -19,6 +19,7 @@ package vm
 import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/math"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
@@ -37,6 +38,7 @@ type Config struct {
 	ParallelTxNum               int                 // Number of slot for transaction execution
 	OptimismPrecompileOverrides PrecompileOverrides // Precompile overrides for Optimism
 	EnableOpcodeOptimizations   bool                // Enable opcode optimization
+	TxDAG                       types.TxDAG
 }
 
 // ScopeContext contains the things that are per-call, such as stack and memory,