diff options
Diffstat (limited to 'eth/fetcher/fetcher.go')
| -rw-r--r-- | eth/fetcher/fetcher.go | 19 |
1 files changed, 19 insertions, 0 deletions
diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 90a202235..07a32b9f1 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -11,6 +11,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" + "github.com/rcrowley/go-metrics" "gopkg.in/karalabe/cookiejar.v2/collections/prque" ) @@ -96,6 +97,13 @@ type Fetcher struct { // Testing hooks fetchingHook func([]common.Hash) // Method to call upon starting a block fetch importedHook func(*types.Block) // Method to call upon successful block import + + // Runtime metrics + announceMeter metrics.Meter // Counter for metering the inbound announcements + announceTimer metrics.Timer // Counter and timer for metering the announce forwarding + broadcastMeter metrics.Meter // Counter for metering the inbound propagations + broadcastTimer metrics.Timer // Counter and timer for metering the block forwarding + discardMeter metrics.Meter // Counter for metering the discarded blocks } // New creates a block fetcher to retrieve blocks based on hash announcements. @@ -118,6 +126,11 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo chainHeight: chainHeight, insertChain: insertChain, dropPeer: dropPeer, + announceMeter: metrics.GetOrRegisterMeter("eth/sync/RemoteAnnounces", metrics.DefaultRegistry), + announceTimer: metrics.GetOrRegisterTimer("eth/sync/LocalAnnounces", metrics.DefaultRegistry), + broadcastMeter: metrics.GetOrRegisterMeter("eth/sync/RemoteBroadcasts", metrics.DefaultRegistry), + broadcastTimer: metrics.GetOrRegisterTimer("eth/sync/LocalBroadcasts", metrics.DefaultRegistry), + discardMeter: metrics.GetOrRegisterMeter("eth/sync/DiscardedBlocks", metrics.DefaultRegistry), } } @@ -229,6 +242,8 @@ func (f *Fetcher) loop() { case notification := <-f.notify: // A block was announced, make sure the peer isn't DOSing us + f.announceMeter.Mark(1) + count := f.announces[notification.origin] + 1 if count > hashLimit { glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, hashLimit) @@ -246,6 +261,7 @@ func (f *Fetcher) loop() { case op := <-f.inject: // A direct block insertion was requested, try and fill any pending gaps + f.broadcastMeter.Mark(1) f.enqueue(op.origin, op.block) case hash := <-f.done: @@ -364,6 +380,7 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { // Discard any past or too distant blocks if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) + f.discardMeter.Mark(1) return } // Schedule the block for future importing @@ -404,6 +421,7 @@ func (f *Fetcher) insert(peer string, block *types.Block) { f.dropPeer(peer) return } + f.broadcastTimer.UpdateSince(block.ReceivedAt) go f.broadcastBlock(block, true) // Run the actual import and log any issues @@ -412,6 +430,7 @@ func (f *Fetcher) insert(peer string, block *types.Block) { return } // If import succeeded, broadcast the block + f.announceTimer.UpdateSince(block.ReceivedAt) go f.broadcastBlock(block, false) // Invoke the testing hook if needed |