eth/downloader: bump block cache target from 256 mb to 2 gb. increase max in-flight block body requests from 8192 to 32768. add a metric for count of in-flight block body retrievals.

eth/downloader/metrics.go

@@ -32,6 +32,9 @@ var (
 	bodyDropMeter    = metrics.NewRegisteredMeter("eth/downloader/bodies/drop", nil)
 	bodyTimeoutMeter = metrics.NewRegisteredMeter("eth/downloader/bodies/timeout", nil)
 
+	blockSizeGauge   = metrics.NewRegisteredGauge("eth/downloader/blocks/size", nil)
+	pendingBodyGauge = metrics.NewRegisteredGauge("eth/downloader/bodies/pending", nil)
+
 	receiptInMeter      = metrics.NewRegisteredMeter("eth/downloader/receipts/in", nil)
 	receiptReqTimer     = metrics.NewRegisteredTimer("eth/downloader/receipts/req", nil)
 	receiptDropMeter    = metrics.NewRegisteredMeter("eth/downloader/receipts/drop", nil)

eth/downloader/queue.go

@@ -25,6 +25,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/prque"
@@ -42,10 +43,10 @@ const (
 )
 
 var (
-	blockCacheMaxItems     = 8192              // Maximum number of blocks to cache before throttling the download
+	blockCacheMaxItems     = 8192 * 4               // Maximum number of blocks to cache before throttling the download
-	blockCacheInitialItems = 2048              // Initial number of blocks to start fetching, before we know the sizes of the blocks
+	blockCacheInitialItems = 2048                   // Initial number of blocks to start fetching, before we know the sizes of the blocks
-	blockCacheMemory       = 256 * 1024 * 1024 // Maximum amount of memory to use for block caching
+	blockCacheMemory       = 2 * 1024 * 1024 * 1024 // Maximum amount of memory to use for block caching
-	blockCacheSizeWeight   = 0.1               // Multiplier to approximate the average block size based on past ones
+	blockCacheSizeWeight   = 0.1                    // Multiplier to approximate the average block size based on past ones
 )
 
 var (
@@ -88,6 +89,18 @@ func newFetchResult(header *types.Header, snapSync bool) *fetchResult {
 	return item
 }
 
+func (f *fetchResult) Size() int {
+	receiptsSize := 0
+	for _, r := range f.Receipts {
+		receiptsSize += int(r.Size())
+	}
+	txsSize := 0
+	for _, t := range f.Transactions {
+		txsSize += int(t.Size())
+	}
+	return common.HashLength*len(f.Uncles) + int(unsafe.Sizeof(types.Withdrawal{})) + int(f.Header.Size()) + receiptsSize + txsSize
+}
+
 // body returns a representation of the fetch result as a types.Body object.
 func (f *fetchResult) body() types.Body {
 	return types.Body{

eth/downloader/resultstore.go

@@ -43,6 +43,8 @@ type resultStore struct {
 	throttleThreshold uint64
 
 	lock sync.RWMutex
+
+	pendingCount int
 }
 
 func newResultStore(size int) *resultStore {
@@ -88,6 +90,7 @@ func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, thro
 	if item == nil {
 		item = newFetchResult(header, snapSync)
 		r.items[index] = item
+		r.pendingCount++
 	}
 	return stale, throttled, item, err
 }
@@ -169,6 +172,18 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	if limit > completed {
 		limit = completed
 	}
+
+	var totalSize int
+	for i := 0; i < limit; i++ {
+		totalSize += r.items[i].Size()
+	}
+	if limit > 0 {
+		blockSizeGauge.Update(int64(totalSize / limit))
+		r.pendingCount -= limit
+	}
+
+	pendingBodyGauge.Update(int64(r.pendingCount))
+
 	results := make([]*fetchResult, limit)
 	copy(results, r.items[:limit])