rework throttle mechanism: calculate an EWMA of the gasUsed/blockSize of recently-completed results. use this in conjunction with avg gas used of in-flight results to estimate the average block size and how many to schedule to fit within the block cache mem limit.

eth/downloader/queue.go

@@ -140,6 +140,7 @@ type queue struct {
 
 	resultCache        *resultStore       // Downloaded but not yet delivered fetch results
 	resultSize         common.StorageSize // Approximate size of a block (exponential moving average)
+	resultGasSizeRatio common.StorageSize // gas used vs block size (bytes) EWMA
 
 	lock   *sync.RWMutex
 	active *sync.Cond
@@ -180,8 +181,7 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) {
 	q.receiptTaskQueue.Reset()
 	q.receiptPendPool = make(map[string]*fetchRequest)
 
-	q.resultCache = newResultStore(blockCacheLimit)
+	q.resultCache = newResultStore(blockCacheLimit, thresholdInitialSize)
-	q.resultCache.SetThrottleThreshold(uint64(thresholdInitialSize))
 }
 
 // Close marks the end of the sync, unblocking Results.
@@ -327,11 +327,12 @@ func (q *queue) Results(block bool) []*fetchResult {
 		size += common.StorageSize(result.Withdrawals.Size())
 		q.resultSize = common.StorageSize(blockCacheSizeWeight)*size +
 			(1-common.StorageSize(blockCacheSizeWeight))*q.resultSize
+
+		// TODO: do we need to keep the resultSize EWMA above?  it's only used in the metrics rn
+		q.resultGasSizeRatio = common.StorageSize(blockCacheSizeWeight)*(common.StorageSize(result.Header.GasUsed)/size) +
+			(1-common.StorageSize(blockCacheSizeWeight))*q.resultGasSizeRatio
 	}
-	// Using the newly calibrated result size, figure out the new throttle limit
+	throttleThreshold := q.resultCache.SetThrottleTarget(q.resultGasSizeRatio)
-	// on the result cache
-	throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize)
-	throttleThreshold = q.resultCache.SetThrottleThreshold(throttleThreshold)
 
 	// With results removed from the cache, wake throttled fetchers
 	for _, ch := range []chan bool{q.blockWakeCh, q.receiptWakeCh} {

eth/downloader/resultstore.go

@@ -18,6 +18,7 @@ package downloader
 
 import (
 	"fmt"
+	"github.com/ethereum/go-ethereum/common"
 	"sync"
 	"sync/atomic"
 
@@ -46,28 +47,49 @@ type resultStore struct {
 	// contained in items.
 	pendingCount int
 
+	// pendingGasUsed tracks the current total gas used of all in-flight bodies
+	pendingGasUsed uint64
+
+	targetGasScheduled uint64
+
 	lock sync.RWMutex
 }
 
-func newResultStore(size int) *resultStore {
+func newResultStore(size, throttleThreshold int) *resultStore {
 	return &resultStore{
 		resultOffset:      0,
 		items:             make([]*fetchResult, size),
-		throttleThreshold: uint64(size),
+		throttleThreshold: uint64(throttleThreshold),
 	}
 }
 
-// SetThrottleThreshold updates the throttling threshold based on the requested
+// SetThrottleTarget updates the throttling threshold based on a targetRatio of gas used / block size,
-// limit and the total queue capacity. It returns the (possibly capped) threshold
+// using that to estimate the average block size based on current in-flight retrievals and returning
-func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 {
+// the throttle threshold:  how many blocks of the estimated size will fit within the block cache
+// capacity.
+func (r *resultStore) SetThrottleTarget(targetRatio common.StorageSize) uint64 {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
-	limit := uint64(len(r.items))
+	if r.pendingCount == 0 {
-	if threshold >= limit {
+		return r.throttleThreshold
-		threshold = limit
 	}
-	r.throttleThreshold = threshold
+	if targetRatio == 0 {
+		targetRatio = 0.1 // TODO: more informed choice of a minimum here?
+	}
+	// use the target ratio to determine
+	// (pendingGasUsed / target) <- estimated total size of collective in-flight retrievals
+	estBlockSize := common.StorageSize(r.pendingGasUsed/uint64(r.pendingCount)) / targetRatio
+
+	// if pending block(s) have 0 gas used, the estimated block size is just a header
+	estBlockSize = max(estBlockSize, 524)
+	targetRetrievalCount := blockCacheMemory / uint64(estBlockSize)
+
+	limit := uint64(len(r.items))
+	if targetRetrievalCount >= limit {
+		targetRetrievalCount = limit
+	}
+	r.throttleThreshold = targetRetrievalCount
 	return r.throttleThreshold
 }
 
@@ -90,9 +112,14 @@ func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, thro
 		return stale, throttled, item, err
 	}
 	if item == nil {
+		if header.GasUsed+r.pendingGasUsed > r.targetGasScheduled {
+			return false, true, nil, nil
+		}
 		item = newFetchResult(header, snapSync)
 		r.items[index] = item
 		r.pendingCount++
+		pendingBodyGauge.Inc(1)
+		r.pendingGasUsed += header.GasUsed
 	}
 	return stale, throttled, item, err
 }
@@ -176,6 +203,9 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	}
 
 	r.pendingCount -= limit
+	for i := 0; i < limit; i++ {
+		r.pendingGasUsed -= r.items[i].Header.GasUsed
+	}
 	pendingBodyGauge.Update(int64(r.pendingCount))
 
 	results := make([]*fetchResult, limit)