eth/downloader: have the resultQueue control its own throttling. it will allow for scheduling new block retrievals until the cumulative gas used of the in-flight retrievals is greater than a target threshold

eth/downloader/queue.go

@@ -44,7 +44,7 @@ const (
 var (
 	blockCacheMaxItems     = 8192              // Maximum number of blocks to cache before throttling the download
 	blockCacheInitialItems = 16                // Initial number of blocks to start fetching, before we know the sizes of the blocks
-	blockCacheMemory       = 1024 * 1024 * 1024 // Maximum amount of memory to use for block caching
+	blockCacheMemory       = 256 * 1024 * 1024 // Maximum amount of memory to use for block caching
 )
 
 var (
@@ -194,7 +194,6 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) {
 	q.receiptPendPool = make(map[string]*fetchRequest)
 
 	q.resultCache = newResultStore(blockCacheLimit)
-	q.resultCache.SetThrottleThreshold(uint64(thresholdInitialSize))
 }
 
 // Close marks the end of the sync, unblocking Results.
@@ -375,14 +374,6 @@ func (q *queue) Results(block bool) []*fetchResult {
 
 	results := q.resultCache.GetCompleted(maxResultsProcess)
 
-	var throttleThreshold uint64
-	if len(results) > 0 {
-		// set a throttle threshhold based on estimated worst-sized block that can be created for a given gas limit.
-		gasLimit := max(results[0].Header.GasLimit, 10)
-		throttleThreshold = min(uint64(blockCacheMemory)/(gasLimit/10), 2048)
-		throttleThreshold = q.resultCache.SetThrottleThreshold(throttleThreshold)
-	}
-
 	// With results removed from the cache, wake throttled fetchers
 	for _, ch := range []chan bool{q.blockWakeCh, q.receiptWakeCh} {
 		select {
@@ -395,7 +386,7 @@ func (q *queue) Results(block bool) []*fetchResult {
 		q.logTime = time.Now()
 
 		info := q.Stats()
-		info = append(info, "throttle", throttleThreshold)
+		info = append(info, "throttle", 0) // TODO: fix this...
 		log.Debug("Downloader queue stats", info...)
 	}
 	return results

eth/downloader/resultstore.go

@@ -35,12 +35,11 @@ type resultStore struct {
 	// *important* : is not safe to use for indexing without checking against length
 	indexIncomplete atomic.Int32
 
-	// throttleThreshold is the limit up to which we _want_ to fill the
-	// results. If blocks are large, we want to limit the results to less
-	// than the number of available slots, and maybe only fill 1024 out of
-	// 8192 possible places. The queue will, at certain times, recalibrate
-	// this index.
-	throttleThreshold uint64
+	// keep track of the total non-blob gas used in the headers we are scheduling.  when we
+	// exceed a threshold, we will throttle preventing additional requests until some current
+	// ones have completed.
+	// TODO: may want to incorporate blob gas into this
+	itemsGasUsed uint64
 
 	lock sync.RWMutex
 }
@@ -49,21 +48,9 @@ func newResultStore(size int) *resultStore {
 	return &resultStore{
 		resultOffset: 0,
 		items:        make([]*fetchResult, size),
-		throttleThreshold: uint64(size),
 	}
 }
 
-// SetThrottleThreshold updates the throttling threshold based on the requested
-// limit and the total queue capacity. It returns the (possibly capped) threshold
-func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-
-	fmt.Printf("set throttle threshold: %d\n", threshold)
-	r.throttleThreshold = min(uint64(len(r.items)), threshold)
-	return r.throttleThreshold
-}
-
 // AddFetch adds a header for body/receipt fetching. This is used when the queue
 // wants to reserve headers for fetching.
 //
@@ -83,6 +70,10 @@ func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, thro
 		return stale, throttled, item, err
 	}
 	if item == nil {
+		if (r.itemsGasUsed+header.GasUsed)/10 > uint64(blockCacheMemory) {
+			return false, true, nil, nil
+		}
+		r.itemsGasUsed += header.GasUsed
 		item = newFetchResult(header, fastSync)
 		r.items[index] = item
 	}
@@ -105,7 +96,14 @@ func (r *resultStore) GetDeliverySlot(headerNumber uint64) (*fetchResult, bool,
 // the index where the result is stored.
 func (r *resultStore) getFetchResult(headerNumber uint64) (item *fetchResult, index int, stale, throttle bool, err error) {
 	index = int(int64(headerNumber) - int64(r.resultOffset))
-	throttle = index >= int(r.throttleThreshold)
+
+	// estimate an average block size based on the worst case:
+	// a block filled with calldata containing all zeroes
+	// costing ~10 gas per byte of calldata.
+	avgEstimatedBlockSize := (r.itemsGasUsed / (uint64(index) + 1)) / 10
+
+	throttleThreshold := min(uint64(len(r.items)), uint64(blockCacheMemory)/avgEstimatedBlockSize)
+	throttle = index >= int(throttleThreshold)
 	stale = index < 0
 
 	if index >= len(r.items) {
@@ -157,6 +155,12 @@ func (r *resultStore) countCompleted() int {
 	return int(index)
 }
 
+/*
+func (r *resultStore) GetThrottleThreshold() int {
+	return (r.itemsGasUsed / len(r.items)) / 10
+}
+*/
+
 // GetCompleted returns the next batch of completed fetchResults
 func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	r.lock.Lock()
@@ -172,6 +176,7 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	// Delete the results from the cache and clear the tail.
 	copy(r.items, r.items[limit:])
 	for i := len(r.items) - limit; i < len(r.items); i++ {
+		r.itemsGasUsed -= r.items[i].Header.GasUsed
 		r.items[i] = nil
 	}
 	// Advance the expected block number of the first cache entry