diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index ce96cf5eeb..bb588ff750 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -370,7 +370,6 @@ func (q *queue) Results(block bool) []*fetchResult { q.lock.Unlock() } // Regardless if closed or not, we can still deliver whatever we have - results := q.resultCache.GetCompleted(maxResultsProcess) // With results removed from the cache, wake throttled fetchers @@ -385,7 +384,7 @@ func (q *queue) Results(block bool) []*fetchResult { q.logTime = time.Now() info := q.Stats() - info = append(info, "throttle", 0) // TODO: fix this... + info = append(info, "throttle", q.resultCache.throttleThreshold()) log.Debug("Downloader queue stats", info...) } return results diff --git a/eth/downloader/resultstore.go b/eth/downloader/resultstore.go index 0d1da6a8d8..46d10a80ec 100644 --- a/eth/downloader/resultstore.go +++ b/eth/downloader/resultstore.go @@ -18,12 +18,21 @@ package downloader import ( "fmt" + "reflect" "sync" "sync/atomic" "github.com/ethereum/go-ethereum/core/types" ) +// returns a rough estimate of the worst-case block size (in bytes) given the +// block gas used. The worst-case block size is assumed to be constructed by +// inserting a transaction with calldata containing all-zeros. With changes +// introduced in EIP-XXX, the price of zero-byte calldata becomes ~10 gas / byte. +func estWorstCaseBlockSize(gasUsed uint64) (size uint64) { + return gasUsed / 10 +} + // resultStore implements a structure for maintaining fetchResults, tracking their // download-progress and delivering (finished) results. type resultStore struct { @@ -91,28 +100,38 @@ func (r *resultStore) GetDeliverySlot(headerNumber uint64) (*fetchResult, bool, return res, stale, err } -// throttleThreshold returns whether the given result index is throttled. -// If the cumulative gas used of all scheduled headers before the given index -// exceeds a threshold, then the index is throttled. -// Cumulative gas used for a set of blocks is used as a proxy for the worst-case -// size of the blocks. Worst-case block size is estimated by assuming that the -// blocks each contain a transaction filled with calldata that is all zeroes. -// The size of a worst-case block is ~= gasUsed / 10 -func (r *resultStore) throttleThreshold(index int) bool { - // estimate the average size of a worst-case block, given what we have already queued: - // total gas used by in flight blocks / number of in-flight blocks / 10 - var headerSize uint64 = 524 - estBlockSize := max((r.itemsGasUsed/(uint64(index)+1))/10, headerSize) +var headerSize uint64 = uint64(reflect.TypeOf(types.Header{}).Size()) - throttleThreshold := min(uint64(len(r.items)), uint64(blockCacheMemory)/estBlockSize+1) - return index >= int(throttleThreshold) +// isThrottled returns whether the given result index is throttled. +func (r *resultStore) isThrottled(index int) bool { + return index >= r.throttleThreshold() +} + +// throttleThreshold returns the index at which block requests will be throttled. +// It is calculated by projecting an average worst-case expected block size for +// in-flight block retrievals: +// +// Take the average gas used per in-flight block +// and estimate the size of the block that could be constructed by filling the +// block with a tx containing all-zero calldata (10 gas per byte of calldata). +// +// From this average worst-case block size, calculate the number of in-flight +// retrievals that can be allowed until the cumulative block size hits a +// predetermined threshold (1 gb). +func (r *resultStore) throttleThreshold() int { + index := r.indexIncomplete.Load() + avgGasUsed := r.itemsGasUsed / (uint64(index) + 1) + blockSize := max(estWorstCaseBlockSize(avgGasUsed), headerSize) + + throttleThreshold := min(uint64(len(r.items)), uint64(blockCacheMemory)/blockSize+1) + return int(throttleThreshold) } // getFetchResult returns the fetchResult corresponding to the given item, and // 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 = r.throttleThreshold(index) + throttle = r.isThrottled(index) stale = index < 0 if index >= len(r.items) { @@ -164,12 +183,6 @@ 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()