Skip to content

Commit

Permalink
core: use less memory during reorgs (ethereum#24616)
Browse files Browse the repository at this point in the history
This PR significantly reduces the memory consumption of a long reorg
  • Loading branch information
MariusVanDerWijden authored and jagdeep sidhu committed May 31, 2022
1 parent c42bcd5 commit abd8282
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 13 deletions.
38 changes: 25 additions & 13 deletions core/blockchain.go
Original file line number Diff line number Diff line change
Expand Up @@ -1966,8 +1966,8 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
oldChain types.Blocks
commonBlock *types.Block

deletedTxs types.Transactions
addedTxs types.Transactions
deletedTxs []common.Hash
addedTxs []common.Hash

deletedLogs [][]*types.Log
rebirthLogs [][]*types.Log
Expand All @@ -1978,7 +1978,9 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
// Old chain is longer, gather all transactions and logs as deleted ones
for ; oldBlock != nil && oldBlock.NumberU64() != newBlock.NumberU64(); oldBlock = bc.GetBlock(oldBlock.ParentHash(), oldBlock.NumberU64()-1) {
oldChain = append(oldChain, oldBlock)
deletedTxs = append(deletedTxs, oldBlock.Transactions()...)
for _, tx := range oldBlock.Transactions() {
deletedTxs = append(deletedTxs, tx.Hash())
}

// Collect deleted logs for notification
logs := bc.collectLogs(oldBlock.Hash(), true)
Expand Down Expand Up @@ -2008,7 +2010,9 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
}
// Remove an old block as well as stash away a new block
oldChain = append(oldChain, oldBlock)
deletedTxs = append(deletedTxs, oldBlock.Transactions()...)
for _, tx := range oldBlock.Transactions() {
deletedTxs = append(deletedTxs, tx.Hash())
}

// Collect deleted logs for notification
logs := bc.collectLogs(oldBlock.Hash(), true)
Expand All @@ -2027,6 +2031,7 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
return fmt.Errorf("invalid new chain")
}
}

// Ensure the user sees large reorgs
if len(oldChain) > 0 && len(newChain) > 0 {
logFn := log.Info
Expand All @@ -2043,7 +2048,7 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
} else if len(newChain) > 0 {
// Special case happens in the post merge stage that current head is
// the ancestor of new head while these two blocks are not consecutive
log.Info("Extend chain", "add", len(newChain), "number", newChain[0].NumberU64(), "hash", newChain[0].Hash())
log.Info("Extend chain", "add", len(newChain), "number", newChain[0].Number(), "hash", newChain[0].Hash())
blockReorgAddMeter.Mark(int64(len(newChain)))
} else {
// len(newChain) == 0 && len(oldChain) > 0
Expand All @@ -2056,19 +2061,17 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
// Insert the block in the canonical way, re-writing history
bc.writeHeadBlock(newChain[i])

// Collect reborn logs due to chain reorg
logs := bc.collectLogs(newChain[i].Hash(), false)
if len(logs) > 0 {
rebirthLogs = append(rebirthLogs, logs)
}
// Collect the new added transactions.
addedTxs = append(addedTxs, newChain[i].Transactions()...)
for _, tx := range newChain[i].Transactions() {
addedTxs = append(addedTxs, tx.Hash())
}
}

// Delete useless indexes right now which includes the non-canonical
// transaction indexes, canonical chain indexes which above the head.
indexesBatch := bc.db.NewBatch()
for _, tx := range types.TxDifference(deletedTxs, addedTxs) {
rawdb.DeleteTxLookupEntry(indexesBatch, tx.Hash())
for _, tx := range types.HashDifference(deletedTxs, addedTxs) {
rawdb.DeleteTxLookupEntry(indexesBatch, tx)
}
// Delete any canonical number assignments above the new head
number := bc.CurrentBlock().NumberU64()
Expand All @@ -2082,6 +2085,15 @@ func (bc *BlockChain) reorg(oldBlock, newBlock *types.Block) error {
if err := indexesBatch.Write(); err != nil {
log.Crit("Failed to delete useless indexes", "err", err)
}

// Collect the logs
for i := len(newChain) - 1; i >= 1; i-- {
// Collect reborn logs due to chain reorg
logs := bc.collectLogs(newChain[i].Hash(), false)
if len(logs) > 0 {
rebirthLogs = append(rebirthLogs, logs)
}
}
// If any logs need to be fired, do it now. In theory we could avoid creating
// this goroutine if there are no events to fire, but realistcally that only
// ever happens if we're reorging empty blocks, which will only happen on idle
Expand Down
18 changes: 18 additions & 0 deletions core/types/transaction.go
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,24 @@ func TxDifference(a, b Transactions) Transactions {
return keep
}

// HashDifference returns a new set which is the difference between a and b.
func HashDifference(a, b []common.Hash) []common.Hash {
keep := make([]common.Hash, 0, len(a))

remove := make(map[common.Hash]struct{})
for _, hash := range b {
remove[hash] = struct{}{}
}

for _, hash := range a {
if _, ok := remove[hash]; !ok {
keep = append(keep, hash)
}
}

return keep
}

// TxByNonce implements the sort interface to allow sorting a list of transactions
// by their nonces. This is usually only useful for sorting transactions from a
// single account, otherwise a nonce comparison doesn't make much sense.
Expand Down

0 comments on commit abd8282

Please sign in to comment.