aboutsummaryrefslogtreecommitdiffstats
path: root/eth/downloader
diff options
context:
space:
mode:
authorPéter Szilágyi <peterke@gmail.com>2015-08-25 18:57:49 +0800
committerPéter Szilágyi <peterke@gmail.com>2015-08-25 22:48:47 +0800
commit17f65cd1e5e0fea6e4f7b96c60767aaa0ada366d (patch)
tree946f0e2478d8002194c09705e638c74b9d3a8ec5 /eth/downloader
parent47a7fe5d22fe2a6be783f6576070814fe951eaaf (diff)
downloadgo-tangerine-17f65cd1e5e0fea6e4f7b96c60767aaa0ada366d.tar.gz
go-tangerine-17f65cd1e5e0fea6e4f7b96c60767aaa0ada366d.tar.zst
go-tangerine-17f65cd1e5e0fea6e4f7b96c60767aaa0ada366d.zip
eth: update metrics collection to handle eth/62 algos
Diffstat (limited to 'eth/downloader')
-rw-r--r--eth/downloader/downloader.go44
-rw-r--r--eth/downloader/metrics.go45
-rw-r--r--eth/downloader/queue.go12
3 files changed, 97 insertions, 4 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 0e8529756..574f2ba15 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -526,6 +526,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error {
glog.V(logger.Debug).Infof("%v: downloading hashes from #%d", p, from)
// Create a timeout timer, and the associated hash fetcher
+ request := time.Now() // time of the last fetch request
timeout := time.NewTimer(0) // timer to dump a non-responsive active peer
<-timeout.C // timeout channel should be initially empty
defer timeout.Stop()
@@ -534,6 +535,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error {
glog.V(logger.Detail).Infof("%v: fetching %d hashes from #%d", p, MaxHashFetch, from)
go p.getAbsHashes(from, MaxHashFetch)
+ request = time.Now()
timeout.Reset(hashTTL)
}
// Start pulling hashes, until all are exhausted
@@ -557,6 +559,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error {
glog.V(logger.Debug).Infof("Received hashes from incorrect peer(%s)", hashPack.peerId)
break
}
+ hashReqTimer.UpdateSince(request)
timeout.Stop()
// If no more hashes are inbound, notify the block fetcher and return
@@ -609,6 +612,7 @@ func (d *Downloader) fetchHashes61(p *peer, td *big.Int, from uint64) error {
case <-timeout.C:
glog.V(logger.Debug).Infof("%v: hash request timed out", p)
+ hashTimeoutMeter.Mark(1)
return errTimeout
}
}
@@ -896,6 +900,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error {
defer glog.V(logger.Debug).Infof("%v: header download terminated", p)
// Create a timeout timer, and the associated hash fetcher
+ request := time.Now() // time of the last fetch request
timeout := time.NewTimer(0) // timer to dump a non-responsive active peer
<-timeout.C // timeout channel should be initially empty
defer timeout.Stop()
@@ -904,6 +909,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error {
glog.V(logger.Detail).Infof("%v: fetching %d headers from #%d", p, MaxHeaderFetch, from)
go p.getAbsHeaders(from, MaxHeaderFetch, 0, false)
+ request = time.Now()
timeout.Reset(headerTTL)
}
// Start pulling headers, until all are exhausted
@@ -927,6 +933,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error {
glog.V(logger.Debug).Infof("Received headers from incorrect peer (%s)", headerPack.peerId)
break
}
+ headerReqTimer.UpdateSince(request)
timeout.Stop()
// If no more headers are inbound, notify the body fetcher and return
@@ -980,6 +987,7 @@ func (d *Downloader) fetchHeaders(p *peer, td *big.Int, from uint64) error {
case <-timeout.C:
// Header retrieval timed out, consider the peer bad and drop
glog.V(logger.Debug).Infof("%v: header request timed out", p)
+ headerTimeoutMeter.Mark(1)
d.dropPeer(p.id)
// Finish the sync gracefully instead of dumping the gathered data though
@@ -1244,7 +1252,14 @@ func (d *Downloader) process() {
// DeliverHashes61 injects a new batch of hashes received from a remote node into
// the download schedule. This is usually invoked through the BlockHashesMsg by
// the protocol handler.
-func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error {
+func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) (err error) {
+ // Update the delivery metrics for both good and failed deliveries
+ hashInMeter.Mark(int64(len(hashes)))
+ defer func() {
+ if err != nil {
+ hashDropMeter.Mark(int64(len(hashes)))
+ }
+ }()
// Make sure the downloader is active
if atomic.LoadInt32(&d.synchronising) == 0 {
return errNoSyncActive
@@ -1265,7 +1280,14 @@ func (d *Downloader) DeliverHashes61(id string, hashes []common.Hash) error {
// DeliverBlocks61 injects a new batch of blocks received from a remote node.
// This is usually invoked through the BlocksMsg by the protocol handler.
-func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error {
+func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) (err error) {
+ // Update the delivery metrics for both good and failed deliveries
+ blockInMeter.Mark(int64(len(blocks)))
+ defer func() {
+ if err != nil {
+ blockDropMeter.Mark(int64(len(blocks)))
+ }
+ }()
// Make sure the downloader is active
if atomic.LoadInt32(&d.synchronising) == 0 {
return errNoSyncActive
@@ -1286,7 +1308,14 @@ func (d *Downloader) DeliverBlocks61(id string, blocks []*types.Block) error {
// DeliverHeaders injects a new batch of blck headers received from a remote
// node into the download schedule.
-func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error {
+func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) (err error) {
+ // Update the delivery metrics for both good and failed deliveries
+ headerInMeter.Mark(int64(len(headers)))
+ defer func() {
+ if err != nil {
+ headerDropMeter.Mark(int64(len(headers)))
+ }
+ }()
// Make sure the downloader is active
if atomic.LoadInt32(&d.synchronising) == 0 {
return errNoSyncActive
@@ -1306,7 +1335,14 @@ func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error {
}
// DeliverBodies injects a new batch of block bodies received from a remote node.
-func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error {
+func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) (err error) {
+ // Update the delivery metrics for both good and failed deliveries
+ bodyInMeter.Mark(int64(len(transactions)))
+ defer func() {
+ if err != nil {
+ bodyDropMeter.Mark(int64(len(transactions)))
+ }
+ }()
// Make sure the downloader is active
if atomic.LoadInt32(&d.synchronising) == 0 {
return errNoSyncActive
diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go
new file mode 100644
index 000000000..fd926affd
--- /dev/null
+++ b/eth/downloader/metrics.go
@@ -0,0 +1,45 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// Contains the metrics collected by the downloader.
+
+package downloader
+
+import (
+ "github.com/ethereum/go-ethereum/metrics"
+)
+
+var (
+ hashInMeter = metrics.NewMeter("eth/downloader/hashes/in")
+ hashReqTimer = metrics.NewTimer("eth/downloader/hashes/req")
+ hashDropMeter = metrics.NewMeter("eth/downloader/hashes/drop")
+ hashTimeoutMeter = metrics.NewMeter("eth/downloader/hashes/timeout")
+
+ blockInMeter = metrics.NewMeter("eth/downloader/blocks/in")
+ blockReqTimer = metrics.NewTimer("eth/downloader/blocks/req")
+ blockDropMeter = metrics.NewMeter("eth/downloader/blocks/drop")
+ blockTimeoutMeter = metrics.NewMeter("eth/downloader/blocks/timeout")
+
+ headerInMeter = metrics.NewMeter("eth/downloader/headers/in")
+ headerReqTimer = metrics.NewTimer("eth/downloader/headers/req")
+ headerDropMeter = metrics.NewMeter("eth/downloader/headers/drop")
+ headerTimeoutMeter = metrics.NewMeter("eth/downloader/headers/timeout")
+
+ bodyInMeter = metrics.NewMeter("eth/downloader/bodies/in")
+ bodyReqTimer = metrics.NewTimer("eth/downloader/bodies/req")
+ bodyDropMeter = metrics.NewMeter("eth/downloader/bodies/drop")
+ bodyTimeoutMeter = metrics.NewMeter("eth/downloader/bodies/timeout")
+)
diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go
index a527414ff..7db78327b 100644
--- a/eth/downloader/queue.go
+++ b/eth/downloader/queue.go
@@ -397,9 +397,19 @@ func (q *queue) Expire(timeout time.Duration) []string {
peers := []string{}
for id, request := range q.pendPool {
if time.Since(request.Time) > timeout {
+ // Update the metrics with the timeout
+ if len(request.Hashes) > 0 {
+ blockTimeoutMeter.Mark(1)
+ } else {
+ bodyTimeoutMeter.Mark(1)
+ }
+ // Return any non satisfied requests to the pool
for hash, index := range request.Hashes {
q.hashQueue.Push(hash, float32(index))
}
+ for _, header := range request.Headers {
+ q.headerQueue.Push(header, -float32(header.Number.Uint64()))
+ }
peers = append(peers, id)
}
}
@@ -420,6 +430,7 @@ func (q *queue) Deliver61(id string, blocks []*types.Block) (err error) {
if request == nil {
return errNoFetchesPending
}
+ blockReqTimer.UpdateSince(request.Time)
delete(q.pendPool, id)
// If no blocks were retrieved, mark them as unavailable for the origin peer
@@ -468,6 +479,7 @@ func (q *queue) Deliver(id string, txLists [][]*types.Transaction, uncleLists []
if request == nil {
return errNoFetchesPending
}
+ bodyReqTimer.UpdateSince(request.Time)
delete(q.pendPool, id)
// If no block bodies were retrieved, mark them as unavailable for the origin peer