remove uncessary memory overhead and reuse SyncPool

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()
 	}
 }
 

core/parallel_state_processor.go

@@ -197,14 +197,7 @@ 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() {
-		for _, slotDB := range stateDBsToRelease {
-			slotDB.PutSyncPool()
-		}
-	}()
 	for _, slot := range p.slotState {
 		slot.pendingTxReqList = make([]*ParallelTxRequest, 0)
 		slot.activatedType = parallelPrimarySlot
@@ -335,7 +328,6 @@ func (p *ParallelStateProcessor) executeInSlot(slotIndex int, txReq *ParallelTxR
 	}
 
 	slotDB.SetTxContext(txReq.tx.Hash(), txReq.txIndex)
-
 	evm, result, err := applyTransactionStageExecution(txReq.msg, gpSlot, slotDB, vmenv, p.delayGasFee)
 	txResult := ParallelTxResult{
 		executedIndex: execNum,
@@ -460,6 +452,8 @@ func (p *ParallelStateProcessor) toConfirmTxIndex(targetTxIndex int, isStage2 bo
 				p.debugConflictRedoNum++
 				// interrupt its current routine, and switch to the other routine
 				p.switchSlot(staticSlotIndex)
+				// reclaim the result.
+				targetResult.slotDB.PutSyncPool()
 				return nil
 			}
 			continue
@@ -794,6 +788,8 @@ func (p *ParallelStateProcessor) confirmTxResults(statedb *state.StateDB, gp *Ga
 		}
 		p.txReqExecuteRecord[resultTxIndex]++
 	}
+	// after merge, the slotDB will not accessible, reclaim the resource
+	result.slotDB.PutSyncPool()
 	return result
 }
 
@@ -949,6 +945,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat
 
 	// clean up when the block is processed
 	p.doCleanUp()
+
 	if p.error != nil {
 		return nil, nil, 0, p.error
 	}

core/state/parallel_statedb.go

@@ -482,6 +482,7 @@ func (s *ParallelStateDB) GetCode(addr common.Address) []byte {
 			code = object.Code()
 		}
 	}
+
 	if _, ok := s.parallel.codeReadsInSlot[addr]; !ok {
 		s.parallel.codeReadsInSlot[addr] = code
 	}

core/state/statedb.go

@@ -187,8 +187,6 @@ type StateDB struct {
 	trieParallelLock        sync.Mutex   // for parallel mode of trie, mostly for get states/objects from trie, lock required to handle trie tracer.
 	stateObjectDestructLock sync.RWMutex // for parallel mode, used in mainDB for mergeSlot and conflict check.
 	snapDestructs           map[common.Address]struct{}
-	snapAccounts            map[common.Address][]byte
-	snapStorage             map[common.Address]map[string][]byte
 
 	// originalRoot is the pre-state root, before any changes were made.
 	// It will be updated when the Commit is called.
@@ -1253,14 +1251,6 @@ var addressToUintPool = sync.Pool{
 	New: func() interface{} { return make(map[common.Address]uint64, defaultNumOfSlots) },
 }
 
-var snapStoragePool = sync.Pool{
-	New: func() interface{} { return make(map[common.Address]map[string][]byte, defaultNumOfSlots) },
-}
-
-var snapStorageValuePool = sync.Pool{
-	New: func() interface{} { return make(map[string][]byte, defaultNumOfSlots) },
-}
-
 var logsPool = sync.Pool{
 	New: func() interface{} { return make(map[common.Hash][]*types.Log, defaultNumOfSlots) },
 }
@@ -1361,20 +1351,6 @@ func (s *StateDB) PutSyncPool() {
 		delete(s.parallel.createdObjectRecord, key)
 	}
 	addressToStructPool.Put(s.parallel.createdObjectRecord)
-
-	for key := range s.snapAccounts {
-		delete(s.snapAccounts, key)
-	}
-	addressToBytesPool.Put(s.snapAccounts)
-
-	for key, storage := range s.snapStorage {
-		for key := range storage {
-			delete(storage, key)
-		}
-		snapStorageValuePool.Put(storage)
-		delete(s.snapStorage, key)
-	}
-	snapStoragePool.Put(s.snapStorage)
 }
 
 func NewEmptySlotDB() *ParallelStateDB {
@@ -2654,10 +2630,6 @@ func (s *StateDB) MergeSlotDB(slotDb *ParallelStateDB, slotReceipt *types.Receip
 				// remove the addr from snapAccounts&snapStorage only when object is deleted.
 				// "deleted" is not equal to "snapDestructs", since createObject() will add an addr for
 				//  snapDestructs to destroy previous object, while it will keep the addr in snapAccounts & snapAccounts
-				s.snapParallelLock.Lock()
-				delete(s.snapAccounts, addr)
-				delete(s.snapStorage, addr)
-				s.snapParallelLock.Unlock()
 				s.AccountMux.Lock()
 				delete(s.accounts, dirtyObj.addrHash)      // Clear out any previously updated account data (may be recreated via a resurrect)
 				delete(s.accountsOrigin, dirtyObj.address) // Clear out any previously updated account data (may be recreated via a resurrect)
@@ -2721,10 +2693,6 @@ func (s *StateDB) MergeSlotDB(slotDb *ParallelStateDB, slotReceipt *types.Receip
 					// remove the addr from snapAccounts&snapStorage only when object is deleted.
 					// "deleted" is not equal to "snapDestructs", since createObject() will add an addr for
 					//  snapDestructs to destroy previous object, while it will keep the addr in snapAccounts & snapAccounts
-					s.snapParallelLock.Lock()
-					delete(s.snapAccounts, addr)
-					delete(s.snapStorage, addr)
-					s.snapParallelLock.Unlock()
 					s.AccountMux.Lock()
 					delete(s.accounts, dirtyObj.addrHash)      // Clear out any previously updated account data (may be recreated via a resurrect)
 					delete(s.accountsOrigin, dirtyObj.address) // Clear out any previously updated account data (may be recreated via a resurrect)
@@ -2847,5 +2815,6 @@ func (m *ParallelDBManager) allocate() *ParallelStateDB {
 
 	elem := m.pool.Front()
 	m.pool.Remove(elem)
-	return elem.Value.(*ParallelStateDB)
+	ret := elem.Value.(*ParallelStateDB)
+	return ret
 }