From f0da3e014c4be94b2500bb11d158cc86e4a064f6 Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Wed, 22 Jan 2020 21:21:44 +0100 Subject: [PATCH] eth/fetcher, eth/handler: differentiate metrics for all/useful broadcasts and announcements --- eth/fetcher/fetcher.go | 53 ++++++++++++++++++++++++------------------ eth/fetcher/metrics.go | 9 +++++-- eth/handler.go | 7 +++--- eth/metrics.go | 5 ++++ 4 files changed, 47 insertions(+), 27 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 28c532d9bd..b0faff3aad 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -46,6 +46,9 @@ var ( // blockRetrievalFn is a callback type for retrieving a block from the local chain. type blockRetrievalFn func(common.Hash) *types.Block +// blockCheckFn is a callback type for checking if a given has exists in the local chain. +type blockCheckFn func(common.Hash) bool + // headerRequesterFn is a callback type for sending a header retrieval request. type headerRequesterFn func(common.Hash) error @@ -130,6 +133,7 @@ type Fetcher struct { // Callbacks getBlock blockRetrievalFn // Retrieves a block from the local chain + hasBlock blockCheckFn // Retrieves a block from the local chain verifyHeader headerVerifierFn // Checks if a block's headers have a valid proof of work broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers chainHeight chainHeightFn // Retrieves the current chain's height @@ -147,21 +151,23 @@ type Fetcher struct { // New creates a block fetcher to retrieve blocks based on hash announcements. func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher { return &Fetcher{ - notify: make(chan *announce), - inject: make(chan *inject), - headerFilter: make(chan chan *headerFilterTask), - bodyFilter: make(chan chan *bodyFilterTask), - done: make(chan common.Hash), - quit: make(chan struct{}), - announces: make(map[string]int), - announced: make(map[common.Hash][]*announce), - fetching: make(map[common.Hash]*announce), - fetched: make(map[common.Hash][]*announce), - completing: make(map[common.Hash]*announce), - queue: prque.New(nil), - queues: make(map[string]int), - queued: make(map[common.Hash]*inject), - getBlock: getBlock, + notify: make(chan *announce), + inject: make(chan *inject), + headerFilter: make(chan chan *headerFilterTask), + bodyFilter: make(chan chan *bodyFilterTask), + done: make(chan common.Hash), + quit: make(chan struct{}), + announces: make(map[string]int), + announced: make(map[common.Hash][]*announce), + fetching: make(map[common.Hash]*announce), + fetched: make(map[common.Hash][]*announce), + completing: make(map[common.Hash]*announce), + queue: prque.New(nil), + queues: make(map[string]int), + queued: make(map[common.Hash]*inject), + getBlock: getBlock, + // This can be replaced this with a more optimized lookup + hasBlock: func(hash common.Hash) bool { return getBlock(hash) != nil }, verifyHeader: verifyHeader, broadcastBlock: broadcastBlock, chainHeight: chainHeight, @@ -194,6 +200,7 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time fetchHeader: headerFetcher, fetchBodies: bodyFetcher, } + propAnnounceUsefulInMeter.Mark(1) select { case f.notify <- block: return nil @@ -204,6 +211,10 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time // Enqueue tries to fill gaps the fetcher's future import queue. func (f *Fetcher) Enqueue(peer string, block *types.Block) error { + if f.hasBlock(block.Hash()) { + return nil + } + propBroadcastUsefulMeter.Mark(1) op := &inject{ origin: peer, block: block, @@ -304,7 +315,7 @@ func (f *Fetcher) loop() { break } // Otherwise if fresh and still unknown, try and import - if number+maxUncleDist < height || f.getBlock(hash) != nil { + if number+maxUncleDist < height || f.hasBlock(hash) { f.forgetBlock(hash) continue } @@ -318,7 +329,6 @@ func (f *Fetcher) loop() { case notification := <-f.notify: // A block was announced, make sure the peer isn't DOSing us - propAnnounceInMeter.Mark(1) count := f.announces[notification.origin] + 1 if count > hashLimit { @@ -352,7 +362,6 @@ func (f *Fetcher) loop() { case op := <-f.inject: // A direct block insertion was requested, try and fill any pending gaps - propBroadcastInMeter.Mark(1) f.enqueue(op.origin, op.block) case hash := <-f.done: @@ -371,7 +380,7 @@ func (f *Fetcher) loop() { f.forgetHash(hash) // If the block still didn't arrive, queue for fetching - if f.getBlock(hash) == nil { + if !f.hasBlock(hash) { request[announce.origin] = append(request[announce.origin], hash) f.fetching[hash] = announce } @@ -406,7 +415,7 @@ func (f *Fetcher) loop() { f.forgetHash(hash) // If the block still didn't arrive, queue for completion - if f.getBlock(hash) == nil { + if !f.hasBlock(hash) { request[announce.origin] = append(request[announce.origin], hash) f.completing[hash] = announce } @@ -453,7 +462,7 @@ func (f *Fetcher) loop() { continue } // Only keep if not imported by other means - if f.getBlock(hash) == nil { + if !f.hasBlock(hash) { announce.header = header announce.time = task.time @@ -527,7 +536,7 @@ func (f *Fetcher) loop() { // Mark the body matched, reassemble if still unknown matched = true - if f.getBlock(hash) == nil { + if !f.hasBlock(hash) { block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i]) block.ReceivedAt = task.time diff --git a/eth/fetcher/metrics.go b/eth/fetcher/metrics.go index d68d12f000..0e1c3ba0ca 100644 --- a/eth/fetcher/metrics.go +++ b/eth/fetcher/metrics.go @@ -23,12 +23,17 @@ import ( ) var ( - propAnnounceInMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/in", nil) + + // Useful block announcements == announcements we did not already know about + propAnnounceUsefulInMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/useful", nil) + propAnnounceOutTimer = metrics.NewRegisteredTimer("eth/fetcher/prop/announces/out", nil) propAnnounceDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/drop", nil) propAnnounceDOSMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/announces/dos", nil) - propBroadcastInMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/in", nil) + // All useful incoming block broadcasts == broadcasts we did not already have + propBroadcastUsefulMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/useful", nil) + propBroadcastOutTimer = metrics.NewRegisteredTimer("eth/fetcher/prop/broadcasts/out", nil) propBroadcastDropMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/drop", nil) propBroadcastDOSMeter = metrics.NewRegisteredMeter("eth/fetcher/prop/broadcasts/dos", nil) diff --git a/eth/handler.go b/eth/handler.go index 2069e4b164..70eb3e3750 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -670,6 +670,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { for _, block := range announces { p.MarkBlock(block.Hash) } + propAnnounceAllInMeter.Mark(int64(len(announces))) // Schedule all the unknown hashes for retrieval unknown := make(newBlockHashesData, 0, len(announces)) for _, block := range announces { @@ -687,6 +688,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&request); err != nil { return errResp(ErrDecode, "%v: %v", msg, err) } + propBroadcastAllInMeter.Mark(1) if hash := types.CalcUncleHash(request.Block.Uncles()); hash != request.Block.UncleHash() { log.Warn("Propagated block has invalid uncles", "have", hash, "exp", request.Block.UncleHash()) break // TODO(karalabe): return error eventually, but wait a few releases @@ -703,9 +705,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Mark the peer as owning the block and schedule it for import p.MarkBlock(request.Block.Hash()) - if !pm.blockchain.HasBlock(request.Block.Hash(), request.Block.NumberU64()) { - pm.fetcher.Enqueue(p.id, request.Block) - } + pm.fetcher.Enqueue(p.id, request.Block) + // Assuming the block is importable by the peer, but possibly not yet done so, // calculate the head hash and TD that the peer truly must have. var ( diff --git a/eth/metrics.go b/eth/metrics.go index 0533a2a875..9a94f3c3e6 100644 --- a/eth/metrics.go +++ b/eth/metrics.go @@ -54,6 +54,11 @@ var ( miscInTrafficMeter = metrics.NewRegisteredMeter("eth/misc/in/traffic", nil) miscOutPacketsMeter = metrics.NewRegisteredMeter("eth/misc/out/packets", nil) miscOutTrafficMeter = metrics.NewRegisteredMeter("eth/misc/out/traffic", nil) + + // All incoming block announcements (no dedup) + propAnnounceAllInMeter = metrics.NewRegisteredMeter("eth/announces/block/in/all", nil) + // All incoming block broadcasts + propBroadcastAllInMeter = metrics.NewRegisteredMeter("eth/broadcasts/block/in/all", nil) ) // meteredMsgReadWriter is a wrapper around a p2p.MsgReadWriter, capable of