make use of stateDBsToRelease

sunny2022da · Sep 2, 2024 · 927abbf · 927abbf
1 parent 4687c11
commit 927abbf
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 4 deletions.
diff --git a/core/blockchain.go b/core/blockchain.go
@@ -1956,8 +1956,10 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 					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)
+					if bc.processor == bc.parallelProcessor {
+						statedb.CreateParallelDBManager(2 * txsCount)
+						log.Debug("Enable 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
@@ -2924,7 +2926,8 @@ func (bc *BlockChain) UseParallelProcessor() {
 		bc.parallelExecution = true
 		bc.processor = bc.parallelProcessor
 	} else {
-		bc.CreateParallelProcessor(bc.vmConfig.ParallelTxNum)
+		log.Error("bc.ParallelProcessor is nil! fallback to serial processor!")
+		bc.UseSerialProcessor()
 	}
 }
 

diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go
@@ -131,6 +131,7 @@ func (p *ParallelStateProcessor) init() {
 	log.Info("Parallel execution mode is enabled", "Parallel Num", p.parallelNum,
 		"CPUNum", runtime.NumCPU())
 	p.txResultChan = make(chan *ParallelTxResult, 20000)
+	p.slotDBsToRelease = make([]*state.ParallelStateDB, 200, 20000)
 	p.stopSlotChan = make(chan struct{}, 1)
 	p.stopConfirmChan = make(chan struct{}, 1)
 	p.stopConfirmStage2Chan = make(chan struct{}, 1)
@@ -197,7 +198,6 @@ func (p *ParallelStateProcessor) resetState(txNum int, statedb *state.StateDB) {
 
 	statedb.PrepareForParallel()
 	p.allTxReqs = make([]*ParallelTxRequest, 0, txNum)
-	p.slotDBsToRelease = make([]*state.ParallelStateDB, 0, txNum)
 
 	stateDBsToRelease := p.slotDBsToRelease
 	go func() {
@@ -420,6 +420,8 @@ func (p *ParallelStateProcessor) toConfirmTxIndex(targetTxIndex int, isStage2 bo
 		}
 
 		valid := p.toConfirmTxIndexResult(targetResult, isStage2)
+		// append result as it is going to be processed or merged.
+		p.slotDBsToRelease = append(p.slotDBsToRelease, targetResult.slotDB)
 		if !valid {
 			staticSlotIndex := targetResult.txReq.staticSlotIndex
 			conflictBase := targetResult.slotDB.BaseTxIndex()