From 1a027994ac59058038801c4a69fc92c1275034a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 15 Oct 2024 13:01:26 +0300 Subject: [PATCH] core, eth, ethstats: simplify chain head events --- core/blockchain.go | 22 +++++++-------- core/blockchain_test.go | 5 ++-- core/chain_indexer.go | 9 +++--- core/events.go | 11 ++++---- core/txindexer.go | 4 +-- core/txpool/txpool.go | 2 +- eth/catalyst/simulated_beacon_test.go | 17 ++++++------ eth/filters/filter_system.go | 2 +- eth/filters/filter_system_test.go | 10 +++---- eth/gasprice/gasprice.go | 4 +-- ethstats/ethstats.go | 40 ++++++++------------------- 11 files changed, 53 insertions(+), 73 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index f7c921fe64..c30386ff8d 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -571,15 +571,14 @@ func (bc *BlockChain) SetHead(head uint64) error { } // Send chain head event to update the transaction pool header := bc.CurrentBlock() - block := bc.GetBlock(header.Hash(), header.Number.Uint64()) - if block == nil { + if block := bc.GetBlock(header.Hash(), header.Number.Uint64()); block == nil { // This should never happen. In practice, previously currentBlock // contained the entire block whereas now only a "marker", so there // is an ever so slight chance for a race we should handle. log.Error("Current block not found in database", "block", header.Number, "hash", header.Hash()) return fmt.Errorf("current block missing: #%d [%x..]", header.Number, header.Hash().Bytes()[:4]) } - bc.chainHeadFeed.Send(ChainHeadEvent{Block: block}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: header}) return nil } @@ -593,15 +592,14 @@ func (bc *BlockChain) SetHeadWithTimestamp(timestamp uint64) error { } // Send chain head event to update the transaction pool header := bc.CurrentBlock() - block := bc.GetBlock(header.Hash(), header.Number.Uint64()) - if block == nil { + if block := bc.GetBlock(header.Hash(), header.Number.Uint64()); block == nil { // This should never happen. In practice, previously currentBlock // contained the entire block whereas now only a "marker", so there // is an ever so slight chance for a race we should handle. log.Error("Current block not found in database", "block", header.Number, "hash", header.Hash()) return fmt.Errorf("current block missing: #%d [%x..]", header.Number, header.Hash().Bytes()[:4]) } - bc.chainHeadFeed.Send(ChainHeadEvent{Block: block}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: header}) return nil } @@ -1552,7 +1550,7 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types // Set new head. bc.writeHeadBlock(block) - bc.chainFeed.Send(ChainEvent{Block: block, Hash: block.Hash(), Logs: logs}) + bc.chainFeed.Send(ChainEvent{Header: block.Header()}) if len(logs) > 0 { bc.logsFeed.Send(logs) } @@ -1562,7 +1560,7 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types // we will fire an accumulated ChainHeadEvent and disable fire // event here. if emitHeadEvent { - bc.chainHeadFeed.Send(ChainHeadEvent{Block: block}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: block.Header()}) } return CanonStatTy, nil } @@ -1627,7 +1625,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness // Fire a single chain head event if we've progressed the chain defer func() { if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() { - bc.chainHeadFeed.Send(ChainHeadEvent{lastCanon}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: lastCanon.Header()}) } }() // Start the parallel header verifier @@ -2329,7 +2327,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error { var deletedLogs []*types.Log for i := len(oldChain) - 1; i >= 0; i-- { // Also send event for blocks removed from the canon chain. - bc.chainSideFeed.Send(ChainSideEvent{Block: oldChain[i]}) + bc.chainSideFeed.Send(ChainSideEvent{Header: oldChain[i].Header()}) // Collect deleted logs for notification if logs := bc.collectLogs(oldChain[i], true); len(logs) > 0 { @@ -2403,11 +2401,11 @@ func (bc *BlockChain) SetCanonical(head *types.Block) (common.Hash, error) { // Emit events logs := bc.collectLogs(head, false) - bc.chainFeed.Send(ChainEvent{Block: head, Hash: head.Hash(), Logs: logs}) + bc.chainFeed.Send(ChainEvent{Header: head.Header()}) if len(logs) > 0 { bc.logsFeed.Send(logs) } - bc.chainHeadFeed.Send(ChainHeadEvent{Block: head}) + bc.chainHeadFeed.Send(ChainHeadEvent{Header: head.Header()}) context := []interface{}{ "number", head.Number(), diff --git a/core/blockchain_test.go b/core/blockchain_test.go index f107b322b6..3ca753799a 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -1385,9 +1385,8 @@ done: for { select { case ev := <-chainSideCh: - block := ev.Block - if _, ok := expectedSideHashes[block.Hash()]; !ok { - t.Errorf("%d: didn't expect %x to be in side chain", i, block.Hash()) + if _, ok := expectedSideHashes[ev.Header.Hash()]; !ok { + t.Errorf("%d: didn't expect %x to be in side chain", i, ev.Header.Hash()) } i++ diff --git a/core/chain_indexer.go b/core/chain_indexer.go index f5fce72588..2865daa1ff 100644 --- a/core/chain_indexer.go +++ b/core/chain_indexer.go @@ -222,20 +222,19 @@ func (c *ChainIndexer) eventLoop(currentHeader *types.Header, events chan ChainH errc <- nil return } - header := ev.Block.Header() - if header.ParentHash != prevHash { + if ev.Header.ParentHash != prevHash { // Reorg to the common ancestor if needed (might not exist in light sync mode, skip reorg then) // TODO(karalabe, zsfelfoldi): This seems a bit brittle, can we detect this case explicitly? if rawdb.ReadCanonicalHash(c.chainDb, prevHeader.Number.Uint64()) != prevHash { - if h := rawdb.FindCommonAncestor(c.chainDb, prevHeader, header); h != nil { + if h := rawdb.FindCommonAncestor(c.chainDb, prevHeader, ev.Header); h != nil { c.newHead(h.Number.Uint64(), true) } } } - c.newHead(header.Number.Uint64(), false) + c.newHead(ev.Header.Number.Uint64(), false) - prevHeader, prevHash = header, header.Hash() + prevHeader, prevHash = ev.Header, ev.Header.Hash() } } } diff --git a/core/events.go b/core/events.go index ac935a137f..ab8b2b5f84 100644 --- a/core/events.go +++ b/core/events.go @@ -17,7 +17,6 @@ package core import ( - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" ) @@ -31,13 +30,13 @@ type NewMinedBlockEvent struct{ Block *types.Block } type RemovedLogsEvent struct{ Logs []*types.Log } type ChainEvent struct { - Block *types.Block - Hash common.Hash - Logs []*types.Log + Header *types.Header } type ChainSideEvent struct { - Block *types.Block + Header *types.Header } -type ChainHeadEvent struct{ Block *types.Block } +type ChainHeadEvent struct { + Header *types.Header +} diff --git a/core/txindexer.go b/core/txindexer.go index 70fe5f3322..b2f2188595 100644 --- a/core/txindexer.go +++ b/core/txindexer.go @@ -151,9 +151,9 @@ func (indexer *txIndexer) loop(chain *BlockChain) { if done == nil { stop = make(chan struct{}) done = make(chan struct{}) - go indexer.run(rawdb.ReadTxIndexTail(indexer.db), head.Block.NumberU64(), stop, done) + go indexer.run(rawdb.ReadTxIndexTail(indexer.db), head.Header.Number.Uint64(), stop, done) } - lastHead = head.Block.NumberU64() + lastHead = head.Header.Number.Uint64() case <-done: stop = nil done = nil diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index be7435247d..5ce69e3763 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -243,7 +243,7 @@ func (p *TxPool) loop(head *types.Header, chain BlockChain) { select { case event := <-newHeadCh: // Chain moved forward, store the head for later consumption - newHead = event.Block.Header() + newHead = event.Header case head := <-resetDone: // Previous reset finished, update the old head and allow a new reset diff --git a/eth/catalyst/simulated_beacon_test.go b/eth/catalyst/simulated_beacon_test.go index 7db1769234..7e9fd7b324 100644 --- a/eth/catalyst/simulated_beacon_test.go +++ b/eth/catalyst/simulated_beacon_test.go @@ -123,16 +123,16 @@ func TestSimulatedBeaconSendWithdrawals(t *testing.T) { timer := time.NewTimer(12 * time.Second) for { select { - case evt := <-chainHeadCh: - for _, includedTx := range evt.Block.Transactions() { + case ev := <-chainHeadCh: + block := ethService.BlockChain().GetBlock(ev.Header.Hash(), ev.Header.Number.Uint64()) + for _, includedTx := range block.Transactions() { includedTxs[includedTx.Hash()] = struct{}{} } - for _, includedWithdrawal := range evt.Block.Withdrawals() { + for _, includedWithdrawal := range block.Withdrawals() { includedWithdrawals = append(includedWithdrawals, includedWithdrawal.Index) } - // ensure all withdrawals/txs included. this will take two blocks b/c number of withdrawals > 10 - if len(includedTxs) == len(txs) && len(includedWithdrawals) == len(withdrawals) && evt.Block.Number().Cmp(big.NewInt(2)) == 0 { + if len(includedTxs) == len(txs) && len(includedWithdrawals) == len(withdrawals) && ev.Header.Number.Cmp(big.NewInt(2)) == 0 { return } case <-timer.C: @@ -186,11 +186,12 @@ func TestOnDemandSpam(t *testing.T) { ) for { select { - case evt := <-chainHeadCh: - for _, itx := range evt.Block.Transactions() { + case ev := <-chainHeadCh: + block := eth.BlockChain().GetBlock(ev.Header.Hash(), ev.Header.Number.Uint64()) + for _, itx := range block.Transactions() { includedTxs[itx.Hash()] = struct{}{} } - for _, iwx := range evt.Block.Withdrawals() { + for _, iwx := range block.Withdrawals() { includedWxs = append(includedWxs, iwx.Index) } // ensure all withdrawals/txs included. this will take two blocks b/c number of withdrawals > 10 diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go index a3a2787a41..86012b3f9a 100644 --- a/eth/filters/filter_system.go +++ b/eth/filters/filter_system.go @@ -391,7 +391,7 @@ func (es *EventSystem) handleTxsEvent(filters filterIndex, ev core.NewTxsEvent) func (es *EventSystem) handleChainEvent(filters filterIndex, ev core.ChainEvent) { for _, f := range filters[BlocksSubscription] { - f.headers <- ev.Block.Header() + f.headers <- ev.Header } } diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 1d52afb282..aec5ee4166 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -200,7 +200,7 @@ func TestBlockSubscription(t *testing.T) { ) for _, blk := range chain { - chainEvents = append(chainEvents, core.ChainEvent{Hash: blk.Hash(), Block: blk}) + chainEvents = append(chainEvents, core.ChainEvent{Header: blk.Header()}) } chan0 := make(chan *types.Header) @@ -213,13 +213,13 @@ func TestBlockSubscription(t *testing.T) { for i1 != len(chainEvents) || i2 != len(chainEvents) { select { case header := <-chan0: - if chainEvents[i1].Hash != header.Hash() { - t.Errorf("sub0 received invalid hash on index %d, want %x, got %x", i1, chainEvents[i1].Hash, header.Hash()) + if chainEvents[i1].Header.Hash() != header.Hash() { + t.Errorf("sub0 received invalid hash on index %d, want %x, got %x", i1, chainEvents[i1].Header.Hash(), header.Hash()) } i1++ case header := <-chan1: - if chainEvents[i2].Hash != header.Hash() { - t.Errorf("sub1 received invalid hash on index %d, want %x, got %x", i2, chainEvents[i2].Hash, header.Hash()) + if chainEvents[i2].Header.Hash() != header.Hash() { + t.Errorf("sub1 received invalid hash on index %d, want %x, got %x", i2, chainEvents[i2].Header.Hash(), header.Hash()) } i2++ } diff --git a/eth/gasprice/gasprice.go b/eth/gasprice/gasprice.go index 19a6c0010a..fe2e4d408a 100644 --- a/eth/gasprice/gasprice.go +++ b/eth/gasprice/gasprice.go @@ -124,10 +124,10 @@ func NewOracle(backend OracleBackend, params Config, startPrice *big.Int) *Oracl go func() { var lastHead common.Hash for ev := range headEvent { - if ev.Block.ParentHash() != lastHead { + if ev.Header.ParentHash != lastHead { cache.Purge() } - lastHead = ev.Block.Hash() + lastHead = ev.Header.Hash() } }() diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index c845db1164..afed5332df 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -219,7 +219,7 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core // Start a goroutine that exhausts the subscriptions to avoid events piling up var ( quitCh = make(chan struct{}) - headCh = make(chan *types.Block, 1) + headCh = make(chan *types.Header, 1) txCh = make(chan struct{}, 1) ) go func() { @@ -231,7 +231,7 @@ func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core // Notify of chain head events, but drop if too frequent case head := <-chainHeadCh: select { - case headCh <- head.Block: + case headCh <- head.Header: default: } @@ -602,9 +602,9 @@ func (s uncleStats) MarshalJSON() ([]byte, error) { } // reportBlock retrieves the current chain head and reports it to the stats server. -func (s *Service) reportBlock(conn *connWrapper, block *types.Block) error { +func (s *Service) reportBlock(conn *connWrapper, header *types.Header) error { // Gather the block details from the header or block chain - details := s.assembleBlockStats(block) + details := s.assembleBlockStats(header) // Short circuit if the block detail is not available. if details == nil { @@ -625,10 +625,9 @@ func (s *Service) reportBlock(conn *connWrapper, block *types.Block) error { // assembleBlockStats retrieves any required metadata to report a single block // and assembles the block stats. If block is nil, the current head is processed. -func (s *Service) assembleBlockStats(block *types.Block) *blockStats { +func (s *Service) assembleBlockStats(header *types.Header) *blockStats { // Gather the block infos from the local blockchain var ( - header *types.Header td *big.Int txs []txStats uncles []*types.Header @@ -638,16 +637,13 @@ func (s *Service) assembleBlockStats(block *types.Block) *blockStats { fullBackend, ok := s.backend.(fullNodeBackend) if ok { // Retrieve current chain head if no block is given. - if block == nil { - head := fullBackend.CurrentBlock() - block, _ = fullBackend.BlockByNumber(context.Background(), rpc.BlockNumber(head.Number.Uint64())) + if header == nil { + header = fullBackend.CurrentBlock() } - // Short circuit if no block is available. It might happen when - // the blockchain is reorging. + block, _ := fullBackend.BlockByNumber(context.Background(), rpc.BlockNumber(header.Number.Uint64())) if block == nil { return nil } - header = block.Header() td = fullBackend.GetTd(context.Background(), header.Hash()) txs = make([]txStats, len(block.Transactions())) @@ -657,15 +653,12 @@ func (s *Service) assembleBlockStats(block *types.Block) *blockStats { uncles = block.Uncles() } else { // Light nodes would need on-demand lookups for transactions/uncles, skip - if block != nil { - header = block.Header() - } else { + if header == nil { header = s.backend.CurrentHeader() } td = s.backend.GetTd(context.Background(), header.Hash()) txs = []txStats{} } - // Assemble and return the block stats author, _ := s.engine.Author(header) @@ -708,19 +701,10 @@ func (s *Service) reportHistory(conn *connWrapper, list []uint64) error { // Gather the batch of blocks to report history := make([]*blockStats, len(indexes)) for i, number := range indexes { - fullBackend, ok := s.backend.(fullNodeBackend) // Retrieve the next block if it's known to us - var block *types.Block - if ok { - block, _ = fullBackend.BlockByNumber(context.Background(), rpc.BlockNumber(number)) // TODO ignore error here ? - } else { - if header, _ := s.backend.HeaderByNumber(context.Background(), rpc.BlockNumber(number)); header != nil { - block = types.NewBlockWithHeader(header) - } - } - // If we do have the block, add to the history and continue - if block != nil { - history[len(history)-1-i] = s.assembleBlockStats(block) + header, _ := s.backend.HeaderByNumber(context.Background(), rpc.BlockNumber(number)) + if header != nil { + history[len(history)-1-i] = s.assembleBlockStats(header) continue } // Ran out of blocks, cut the report short and send