From 52221ae8cf9cec0d598dd584fccf08931c05ff9b Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Wed, 23 Apr 2025 15:59:48 +0800 Subject: [PATCH] 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 | 17 +++++++------ eth/downloader/resultstore.go | 48 ++++++++++++++++++++++++++++------- 2 files changed, 48 insertions(+), 17 deletions(-) diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index da619e764c..8a117f1f7d 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -138,8 +138,9 @@ type queue struct { receiptPendPool map[string]*fetchRequest // Currently pending receipt retrieval operations receiptWakeCh chan bool // Channel to notify when receipt fetcher of new tasks - resultCache *resultStore // Downloaded but not yet delivered fetch results - resultSize common.StorageSize // Approximate size of a block (exponential moving average) + 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.SetThrottleThreshold(uint64(thresholdInitialSize)) + q.resultCache = newResultStore(blockCacheLimit, 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 - // on the result cache - throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize) - throttleThreshold = q.resultCache.SetThrottleThreshold(throttleThreshold) + throttleThreshold := q.resultCache.SetThrottleTarget(q.resultGasSizeRatio) // With results removed from the cache, wake throttled fetchers for _, ch := range []chan bool{q.blockWakeCh, q.receiptWakeCh} { diff --git a/eth/downloader/resultstore.go b/eth/downloader/resultstore.go index 36863cfde3..f9bad24811 100644 --- a/eth/downloader/resultstore.go +++ b/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 -// limit and the total queue capacity. It returns the (possibly capped) threshold -func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 { +// SetThrottleTarget updates the throttling threshold based on a targetRatio of gas used / block size, +// using that to estimate the average block size based on current in-flight retrievals and returning +// 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 threshold >= limit { - threshold = limit + if r.pendingCount == 0 { + return r.throttleThreshold } - 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)