mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-22 20:56:42 +00:00
core: cache raw receipts of new heads separately
This commit is contained in:
parent
0c5200bb43
commit
444efa9cf5
3 changed files with 31 additions and 36 deletions
|
|
@ -1616,24 +1616,8 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
|
||||||
blockBatch := bc.db.NewBatch()
|
blockBatch := bc.db.NewBatch()
|
||||||
rawdb.WriteBlock(blockBatch, block)
|
rawdb.WriteBlock(blockBatch, block)
|
||||||
blockHash := block.Hash()
|
blockHash := block.Hash()
|
||||||
bc.blockCache.Add(blockHash, block)
|
|
||||||
rawdb.WriteReceipts(blockBatch, blockHash, block.NumberU64(), receipts)
|
rawdb.WriteReceipts(blockBatch, blockHash, block.NumberU64(), receipts)
|
||||||
if receipts != nil {
|
bc.indexServers.cacheRawReceipts(blockHash, receipts)
|
||||||
// ensure that receipts are cached in the exact same format as when reconstructed from database.
|
|
||||||
for _, receipt := range receipts {
|
|
||||||
if receipt.Logs == nil {
|
|
||||||
receipt.Logs = []*types.Log{}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
var blobGasPrice *big.Int
|
|
||||||
if block.ExcessBlobGas() != nil {
|
|
||||||
blobGasPrice = eip4844.CalcBlobFee(bc.chainConfig, block.Header())
|
|
||||||
}
|
|
||||||
types.Receipts(receipts).DeriveFields(bc.chainConfig, blockHash, block.NumberU64(), block.Time(), block.BaseFee(), blobGasPrice, block.Transactions())
|
|
||||||
bc.receiptsCache.Add(blockHash, receipts)
|
|
||||||
} else {
|
|
||||||
bc.receiptsCache.Add(blockHash, []*types.Receipt{})
|
|
||||||
}
|
|
||||||
rawdb.WritePreimages(blockBatch, statedb.Preimages())
|
rawdb.WritePreimages(blockBatch, statedb.Preimages())
|
||||||
if err := blockBatch.Write(); err != nil {
|
if err := blockBatch.Write(); err != nil {
|
||||||
log.Crit("Failed to write block into disk", "err", err)
|
log.Crit("Failed to write block into disk", "err", err)
|
||||||
|
|
|
||||||
|
|
@ -266,14 +266,6 @@ func (bc *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts {
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return bc.GetReceipts(hash, number)
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetReceipts retrieves the receipts for all transactions in a given block.
|
|
||||||
func (bc *BlockChain) GetReceipts(hash common.Hash, number uint64) types.Receipts {
|
|
||||||
if receipts, ok := bc.receiptsCache.Get(hash); ok {
|
|
||||||
return receipts
|
|
||||||
}
|
|
||||||
header := bc.GetHeader(hash, number)
|
header := bc.GetHeader(hash, number)
|
||||||
if header == nil {
|
if header == nil {
|
||||||
return nil
|
return nil
|
||||||
|
|
|
||||||
|
|
@ -22,6 +22,7 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/common"
|
"github.com/ethereum/go-ethereum/common"
|
||||||
|
"github.com/ethereum/go-ethereum/common/lru"
|
||||||
"github.com/ethereum/go-ethereum/core/types"
|
"github.com/ethereum/go-ethereum/core/types"
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
@ -29,6 +30,7 @@ import (
|
||||||
const (
|
const (
|
||||||
busyDelay = time.Second // indexer status polling frequency when not ready
|
busyDelay = time.Second // indexer status polling frequency when not ready
|
||||||
maxHistoricPrefetch = 16 // size of block data pre-fetch queue
|
maxHistoricPrefetch = 16 // size of block data pre-fetch queue
|
||||||
|
rawReceiptsCacheSize = 8
|
||||||
logFrequency = time.Second * 20 // log info frequency during long indexing/unindexing process
|
logFrequency = time.Second * 20 // log info frequency during long indexing/unindexing process
|
||||||
headLogDelay = time.Second // head indexing log info delay (do not log if finished faster)
|
headLogDelay = time.Second // head indexing log info delay (do not log if finished faster)
|
||||||
)
|
)
|
||||||
|
|
@ -79,6 +81,7 @@ type indexServers struct {
|
||||||
lock sync.Mutex
|
lock sync.Mutex
|
||||||
servers []*indexServer
|
servers []*indexServer
|
||||||
chain *BlockChain
|
chain *BlockChain
|
||||||
|
rawReceiptsCache *lru.Cache[common.Hash, []*types.Receipt]
|
||||||
|
|
||||||
lastHead *types.Header
|
lastHead *types.Header
|
||||||
lastHeadReceipts types.Receipts
|
lastHeadReceipts types.Receipts
|
||||||
|
|
@ -96,6 +99,7 @@ func (f *indexServers) init(chain *BlockChain) {
|
||||||
f.chain = chain
|
f.chain = chain
|
||||||
f.lastHead = chain.CurrentBlock()
|
f.lastHead = chain.CurrentBlock()
|
||||||
f.closeCh = make(chan struct{})
|
f.closeCh = make(chan struct{})
|
||||||
|
f.rawReceiptsCache = lru.NewCache[common.Hash, []*types.Receipt](rawReceiptsCacheSize)
|
||||||
}
|
}
|
||||||
|
|
||||||
// stop shuts down all registered Indexers and their serving goroutines.
|
// stop shuts down all registered Indexers and their serving goroutines.
|
||||||
|
|
@ -135,16 +139,25 @@ func (f *indexServers) register(indexer Indexer, name string) {
|
||||||
go server.historicSendLoop()
|
go server.historicSendLoop()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (f *indexServers) cacheRawReceipts(blockHash common.Hash, blockReceipts types.Receipts) {
|
||||||
|
f.rawReceiptsCache.Add(blockHash, blockReceipts)
|
||||||
|
}
|
||||||
|
|
||||||
// broadcast sends a new head block to all registered Indexer instances.
|
// broadcast sends a new head block to all registered Indexer instances.
|
||||||
func (f *indexServers) broadcast(header *types.Header) {
|
func (f *indexServers) broadcast(header *types.Header) {
|
||||||
f.lock.Lock()
|
f.lock.Lock()
|
||||||
defer f.lock.Unlock()
|
defer f.lock.Unlock()
|
||||||
|
|
||||||
blockReceipts := f.chain.GetReceipts(header.Hash(), header.Number.Uint64())
|
blockHash := header.Hash()
|
||||||
|
blockReceipts, _ := f.rawReceiptsCache.Get(blockHash)
|
||||||
|
if blockReceipts == nil {
|
||||||
|
blockReceipts = f.chain.GetRawReceipts(blockHash, header.Number.Uint64())
|
||||||
if blockReceipts == nil {
|
if blockReceipts == nil {
|
||||||
log.Error("Receipts belonging to new head are missing", "number", header.Number, "hash", header.Hash())
|
log.Error("Receipts belonging to new head are missing", "number", header.Number, "hash", header.Hash())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
f.rawReceiptsCache.Add(blockHash, blockReceipts)
|
||||||
|
}
|
||||||
f.lastHead, f.lastHeadReceipts = header, blockReceipts
|
f.lastHead, f.lastHeadReceipts = header, blockReceipts
|
||||||
for _, server := range f.servers {
|
for _, server := range f.servers {
|
||||||
server.sendHeadBlockData(header, blockReceipts)
|
server.sendHeadBlockData(header, blockReceipts)
|
||||||
|
|
@ -499,7 +512,13 @@ func (s *indexServer) historicReadLoop() {
|
||||||
// Send next item to the queue.
|
// Send next item to the queue.
|
||||||
bd := blockData{blockNumber: sendRange.First(), revertCount: status.revertCount}
|
bd := blockData{blockNumber: sendRange.First(), revertCount: status.revertCount}
|
||||||
if bd.header = s.parent.chain.GetHeaderByNumber(bd.blockNumber); bd.header != nil {
|
if bd.header = s.parent.chain.GetHeaderByNumber(bd.blockNumber); bd.header != nil {
|
||||||
bd.receipts = s.parent.chain.GetReceipts(bd.header.Hash(), bd.blockNumber)
|
blockHash := bd.header.Hash()
|
||||||
|
bd.receipts, _ = s.parent.rawReceiptsCache.Get(blockHash)
|
||||||
|
if bd.receipts == nil {
|
||||||
|
bd.receipts = s.parent.chain.GetRawReceipts(blockHash, bd.blockNumber)
|
||||||
|
// Note: we do not cache historical receipts because typically
|
||||||
|
// each indexer requests them at different times.
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// Note that a response with missing block data is still sent in case of
|
// Note that a response with missing block data is still sent in case of
|
||||||
// a read error, signaling to the sender logic that something is missing.
|
// a read error, signaling to the sender logic that something is missing.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue