downloader, fetcher: throttle-metrics, fetcher filter improvements, standalone resultcache

core/types/block.go

@@ -136,6 +136,17 @@ func rlpHash(x interface{}) (h common.Hash) {
 	return h
 }
 
+// EmptyBody returns true if there is no additional 'body' to complete the header
+// that is: no transactions and no uncles
+func (h *Header) EmptyBody() bool{
+	return h.TxHash == EmptyRootHash && h.UncleHash == EmptyUncleHash
+}
+
+// EmptyReceipts returns true if there are no receipts for this header/block
+func (h *Header) EmptyReceipts() bool{
+	return h.ReceiptHash == EmptyRootHash
+}
+
 // Body is a simple (mutable, non-safe) data container for storing and moving
 // a block's data contents (transactions and uncles) together.
 type Body struct {

eth/downloader/downloader.go

@@ -219,7 +219,7 @@ func New(checkpoint uint64, stateDb ethdb.Database, stateBloom *trie.SyncBloom,
 		stateBloom:     stateBloom,
 		mux:            mux,
 		checkpoint:     checkpoint,
-		queue:          newQueue(),
+		queue:          newQueue(blockCacheItems),
 		peers:          newPeerSet(),
 		rttEstimate:    uint64(rttMaxEstimate),
 		rttConfidence:  uint64(1000000),
@@ -619,7 +619,7 @@ func (d *Downloader) fetchHeight(p *peerConnection) (*types.Header, error) {
 			// Make sure the peer actually gave something valid
 			headers := packet.(*headerPack).headers
 			if len(headers) != 1 {
-				p.log.Debug("Multiple headers for single request", "headers", len(headers))
+				p.log.Info("Multiple headers for single request", "headers", len(headers))
 				return nil, errBadPeer
 			}
 			head := headers[0]
@@ -851,7 +851,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header)
 				// Make sure the peer actually gave something valid
 				headers := packer.(*headerPack).headers
 				if len(headers) != 1 {
-					p.log.Debug("Multiple headers for single request", "headers", len(headers))
+					p.log.Info("Multiple headers for single request", "headers", len(headers))
 					return 0, errBadPeer
 				}
 				arrived = true
@@ -875,7 +875,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header)
 				}
 				header := d.lightchain.GetHeaderByHash(h) // Independent of sync mode, header surely exists
 				if header.Number.Uint64() != check {
-					p.log.Debug("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check)
+					p.log.Info("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check)
 					return 0, errBadPeer
 				}
 				start = check
@@ -1120,7 +1120,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
 			pack := packet.(*bodyPack)
 			return d.queue.DeliverBodies(pack.peerID, pack.transactions, pack.uncles)
 		}
-		expire   = func() map[string]int { return d.queue.ExpireBodies(d.requestTTL()) }
+		expire   = func() map[string]int {return d.queue.ExpireBodies(d.requestTTL())}
 		fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchBodies(req) }
 		capacity = func(p *peerConnection) int { return p.BlockCapacity(d.requestRTT()) }
 		setIdle  = func(p *peerConnection, accepted int) { p.SetBodiesIdle(accepted) }
@@ -1188,7 +1188,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 	idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int), kind string) error {
 
 	// Create a ticker to detect expired retrieval tasks
-	ticker := time.NewTicker(100 * time.Millisecond)
+	ticker := time.NewTicker(200 * time.Millisecond)
 	defer ticker.Stop()
 
 	update := make(chan struct{}, 1)
@@ -1264,7 +1264,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 					// The reason the minimum threshold is 2 is because the downloader tries to estimate the bandwidth
 					// and latency of a peer separately, which requires pushing the measures capacity a bit and seeing
 					// how response times reacts, to it always requests one more than the minimum (i.e. min 2).
-					if fails > 2 {
+					if fails > 8 {
 						peer.log.Trace("Data delivery timed out", "type", kind)
 						setIdle(peer, 0)
 					} else {
@@ -1323,6 +1323,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 					progressed = true
 				}
 				if request == nil {
+					//peer.log.Info("no request allocated this loop", "type", kind)
 					continue
 				}
 				if request.From > 0 {
@@ -1359,6 +1360,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) error {
 	// Keep a count of uncertain headers to roll back
 	var rollback []*types.Header
+	var rollbackErr error
 	defer func() {
 		if len(rollback) > 0 {
 			// Flatten the headers and roll them back
@@ -1380,7 +1382,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er
 			log.Warn("Rolled back headers", "count", len(hashes),
 				"header", fmt.Sprintf("%d->%d", lastHeader, d.lightchain.CurrentHeader().Number),
 				"fast", fmt.Sprintf("%d->%d", lastFastBlock, curFastBlock),
-				"block", fmt.Sprintf("%d->%d", lastBlock, curBlock))
+				"block", fmt.Sprintf("%d->%d", lastBlock, curBlock), "reason", rollbackErr)
 		}
 	}()
 
@@ -1469,9 +1471,10 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er
 					if n, err := d.lightchain.InsertHeaderChain(chunk, frequency); err != nil {
 						// If some headers were inserted, add them too to the rollback list
 						if n > 0 {
+							rollbackErr = err
 							rollback = append(rollback, chunk[:n]...)
 						}
-						log.Debug("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err)
+						log.Info("Invalid header encountered", "number", chunk[n].Number, "hash", chunk[n].Hash(), "err", err)
 						return errInvalidChain
 					}
 					// All verifications passed, store newly found uncertain headers
@@ -1493,7 +1496,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er
 					// Otherwise insert the headers for content retrieval
 					inserts := d.queue.Schedule(chunk, origin)
 					if len(inserts) != len(chunk) {
-						log.Debug("Stale headers")
+						rollbackErr = fmt.Errorf("stale headers: len inserts %v len(chunk) %v", len(inserts), len(chunk))
 						return errBadPeer
 					}
 				}
@@ -1693,7 +1696,7 @@ func (d *Downloader) commitFastSyncData(results []*fetchResult, stateSync *state
 	}
 	// Retrieve the a batch of results to import
 	first, last := results[0].Header, results[len(results)-1].Header
-	log.Debug("Inserting fast-sync blocks", "items", len(results),
+	log.Info("Inserting fast-sync blocks", "items", len(results),
 		"firstnum", first.Number, "firsthash", first.Hash(),
 		"lastnumn", last.Number, "lasthash", last.Hash(),
 	)

eth/downloader/metrics.go

@@ -40,4 +40,7 @@ var (
 
 	stateInMeter   = metrics.NewRegisteredMeter("eth/downloader/states/in", nil)
 	stateDropMeter = metrics.NewRegisteredMeter("eth/downloader/states/drop", nil)
+
+	throttleBlockCounter   = metrics.NewRegisteredCounter("eth/downloader/throttle/blocks", nil)
+	throttleReceiptCounter = metrics.NewRegisteredCounter("eth/downloader/throttle/receipts", nil)
 )

eth/downloader/peer.go

@@ -117,11 +117,13 @@ func newPeerConnection(id string, version int, peer Peer, logger log.Logger) *pe
 	return &peerConnection{
 		id:                id,
 		lacking:           make(map[common.Hash]struct{}),
-
 		peer:              peer,
-
 		version:           version,
 		log:               logger,
+		headerThroughput:  float64(MaxHeaderFetch / 8),
+		blockThroughput:   float64(MaxBlockFetch / 8),
+		receiptThroughput: float64(MaxReceiptFetch / 8),
+		stateThroughput:   float64(MaxStateFetch / 8),
 	}
 }
 
@@ -135,10 +137,10 @@ func (p *peerConnection) Reset() {
 	atomic.StoreInt32(&p.receiptIdle, 0)
 	atomic.StoreInt32(&p.stateIdle, 0)
 
-	p.headerThroughput = 0
+	p.headerThroughput = float64(MaxHeaderFetch / 8)
-	p.blockThroughput = 0
+	p.blockThroughput = float64(MaxBlockFetch / 8)
-	p.receiptThroughput = 0
+	p.receiptThroughput = float64(MaxReceiptFetch / 8)
-	p.stateThroughput = 0
+	p.stateThroughput = float64(MaxStateFetch / 8)
 
 	p.lacking = make(map[common.Hash]struct{})
 }
@@ -283,7 +285,7 @@ func (p *peerConnection) HeaderCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	return int(math.Min(1+math.Max(1, p.headerThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxHeaderFetch)))
+	return int(math.Min(7+math.Max(1, p.headerThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxHeaderFetch)))
 }
 
 // BlockCapacity retrieves the peers block download allowance based on its
@@ -292,7 +294,7 @@ func (p *peerConnection) BlockCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	return int(math.Min(1+math.Max(1, p.blockThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxBlockFetch)))
+	return int(math.Min(7+math.Max(1, p.blockThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxBlockFetch)))
 }
 
 // ReceiptCapacity retrieves the peers receipt download allowance based on its
@@ -301,7 +303,7 @@ func (p *peerConnection) ReceiptCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	return int(math.Min(1+math.Max(1, p.receiptThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxReceiptFetch)))
+	return int(math.Min(7+math.Max(1, p.receiptThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxReceiptFetch)))
 }
 
 // NodeDataCapacity retrieves the peers state download allowance based on its
@@ -310,7 +312,7 @@ func (p *peerConnection) NodeDataCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	return int(math.Min(1+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch)))
+	return int(math.Min(7+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch)))
 }
 
 // MarkLacking appends a new entity to the set of items (blocks, receipts, states)
@@ -473,6 +475,12 @@ func (ps *peerSet) HeaderIdlePeers() ([]*peerConnection, int) {
 	return ps.idlePeers(62, 65, idle, throughput)
 }
 
+func fullyIdle(p *peerConnection) bool {
+	return atomic.LoadInt32(&p.blockIdle) == 0 &&
+		atomic.LoadInt32(&p.receiptIdle) == 0 &&
+		atomic.LoadInt32(&p.stateIdle) == 0
+}
+
 // BodyIdlePeers retrieves a flat list of all the currently body-idle peers within
 // the active peer set, ordered by their reputation.
 func (ps *peerSet) BodyIdlePeers() ([]*peerConnection, int) {
@@ -523,22 +531,20 @@ func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peerC
 	defer ps.lock.RUnlock()
 
 	idle, total := make([]*peerConnection, 0, len(ps.peers)), 0
+	tps := make([]float64, 0, len(ps.peers))
 	for _, p := range ps.peers {
 		if p.version >= minProtocol && p.version <= maxProtocol {
 			if idleCheck(p) {
 				idle = append(idle, p)
+				tps = append(tps, throughput(p))
 			}
 			total++
 		}
 	}
-	for i := 0; i < len(idle); i++ {
+	// And sort them
-		for j := i + 1; j < len(idle); j++ {
+	sortPeers := &peerThroughputSort{idle, tps}
-			if throughput(idle[i]) < throughput(idle[j]) {
+	sort.Sort(sortPeers)
-				idle[i], idle[j] = idle[j], idle[i]
+	return sortPeers.p, total
-			}
-		}
-	}
-	return idle, total
 }
 
 // medianRTT returns the median RTT of the peerset, considering only the tuning
@@ -571,3 +577,24 @@ func (ps *peerSet) medianRTT() time.Duration {
 	}
 	return median
 }
+
+// peerThroughputSort implements the Sort interface, and allows for
+// sorting a set of peers by their throughput
+// The sorted data is with the _highest_ throughput first
+type peerThroughputSort struct {
+	p  []*peerConnection
+	tp []float64
+}
+
+func (ps *peerThroughputSort) Len() int {
+	return len(ps.p)
+}
+
+func (ps *peerThroughputSort) Less(i, j int) bool {
+	return ps.tp[i] > ps.tp[j]
+}
+
+func (ps *peerThroughputSort) Swap(i, j int) {
+	ps.p[i], ps.p[j] = ps.p[j], ps.p[i]
+	ps.tp[i], ps.tp[j] = ps.tp[j], ps.tp[i]
+}

eth/downloader/peer_test.go

@@ -0,0 +1,37 @@
+package downloader
+
+import (
+	"sort"
+	"testing"
+)
+
+func TestPeerThroughputSorting(t *testing.T){
+	a := &peerConnection{
+		id:"a",
+		headerThroughput:1.25,
+	}
+	b := &peerConnection{
+		id: "b",
+		headerThroughput:1.21,
+	}
+	c := &peerConnection{
+		id: "c",
+		headerThroughput:1.23,
+	}
+
+	peers := []*peerConnection{a,b,c}
+	tps := []float64{a.headerThroughput,
+	b.headerThroughput, c.headerThroughput}
+	sortPeers := &peerThroughputSort{peers, tps}
+	sort.Sort(sortPeers)
+	if got, exp := sortPeers.p[0].id , "a"; got != exp{
+		t.Errorf("sort fail, got %v exp %v", got, exp)
+	}
+	if got, exp := sortPeers.p[1].id , "c"; got != exp{
+		t.Errorf("sort fail, got %v exp %v", got, exp)
+	}
+	if got, exp := sortPeers.p[2].id , "b"; got != exp{
+		t.Errorf("sort fail, got %v exp %v", got, exp)
+	}
+
+}

eth/downloader/queue.go

@@ -54,7 +54,7 @@ type fetchRequest struct {
 // fetchResult is a struct collecting partial results from data fetchers until
 // all outstanding pieces complete and the result as a whole can be processed.
 type fetchResult struct {
-	Pending int         // Number of data fetches still pending
+	Pending uint8       // Number of data fetches still pending
 	Hash    common.Hash // Hash of the header to prevent recalculating
 
 	Header       *types.Header
@@ -89,8 +89,8 @@ type queue struct {
 	receiptPendPool  map[string]*fetchRequest      // [eth/63] Currently pending receipt retrieval operations
 	receiptDonePool  map[common.Hash]struct{}      // [eth/63] Set of the completed receipt fetches
 
-	resultCache  []*fetchResult     // Downloaded but not yet delivered fetch results
+	resultCache *resultStore // Downloaded but not yet delivered fetch results
-	resultOffset uint64             // Offset of the first cached fetch result in the block chain
+	//resultOffset uint64             // Offset of the first cached fetch result in the block chain
 	resultSize common.StorageSize // Approximate size of a block (exponential moving average)
 
 	lock   *sync.RWMutex
@@ -99,7 +99,7 @@ type queue struct {
 }
 
 // newQueue creates a new download queue for scheduling block retrieval.
-func newQueue() *queue {
+func newQueue(blockCacheLimit int) *queue {
 	lock := new(sync.RWMutex)
 	return &queue{
 		headerPendPool:   make(map[string]*fetchRequest),
@@ -112,7 +112,8 @@ func newQueue() *queue {
 		receiptTaskQueue: prque.New(nil),
 		receiptPendPool:  make(map[string]*fetchRequest),
 		receiptDonePool:  make(map[common.Hash]struct{}),
-		resultCache:      make([]*fetchResult, blockCacheItems),
+		//resultCache:      make([]*fetchResult, blockCacheItems),
+		resultCache: newResultStore(blockCacheLimit),
 		active:      sync.NewCond(lock),
 		lock:        lock,
 	}
@@ -139,8 +140,7 @@ func (q *queue) Reset() {
 	q.receiptPendPool = make(map[string]*fetchRequest)
 	q.receiptDonePool = make(map[common.Hash]struct{})
 
-	q.resultCache = make([]*fetchResult, blockCacheItems)
+	q.resultCache = newResultStore(blockCacheItems)
-	q.resultOffset = 0
 }
 
 // Close marks the end of the sync, unblocking Results.
@@ -219,18 +219,24 @@ func (q *queue) Idle() bool {
 // fetches exceed block cache).
 func (q *queue) ShouldThrottleBlocks() bool {
 	q.lock.Lock()
-	defer q.lock.Unlock()
+	t := q.resultSlots(q.blockPendPool, q.blockDonePool) <= 0
-
+	q.lock.Unlock()
-	return q.resultSlots(q.blockPendPool, q.blockDonePool) <= 0
+	if t {
+		throttleBlockCounter.Inc(1)
+	}
+	return t
 }
 
 // ShouldThrottleReceipts checks if the download should be throttled (active receipt
 // fetches exceed block cache).
 func (q *queue) ShouldThrottleReceipts() bool {
 	q.lock.Lock()
-	defer q.lock.Unlock()
+	t := q.resultSlots(q.receiptPendPool, q.receiptDonePool) <= 0
-
+	q.lock.Unlock()
-	return q.resultSlots(q.receiptPendPool, q.receiptDonePool) <= 0
+	if t {
+		throttleReceiptCounter.Inc(1)
+	}
+	return t
 }
 
 // resultSlots calculates the number of results slots available for requests
@@ -238,30 +244,19 @@ func (q *queue) ShouldThrottleReceipts() bool {
 // cache.
 func (q *queue) resultSlots(pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}) int {
 	// Calculate the maximum length capped by the memory limit
-	limit := len(q.resultCache)
+	cacheItems := len(q.resultCache.items)
-	if common.StorageSize(len(q.resultCache))*q.resultSize > common.StorageSize(blockCacheMemory) {
+	limit := cacheItems
+	if common.StorageSize(cacheItems)*q.resultSize > common.StorageSize(blockCacheMemory) {
 		limit = int((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize)
 	}
 	// Calculate the number of slots already finished
-	finished := 0
+	finished := q.resultCache.countCompleted()
-	for _, result := range q.resultCache[:limit] {
-		if result == nil {
-			break
-		}
-		if _, ok := donePool[result.Hash]; ok {
-			finished++
-		}
-	}
 	// Calculate the number of slots currently downloading
 	pending := 0
+	//iterations := 0
 	for _, request := range pendPool {
-		for _, header := range request.Headers {
+		pending += len(request.Headers)
-			if header.Number.Uint64() < q.resultOffset+uint64(limit) {
-				pending++
 	}
-		}
-	}
-	// Return the free slots to distribute
 	return limit - finished - pending
 }
 
@@ -310,6 +305,8 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
+	// if the resultCache pushes back, we can stop trying to shove things in there for now
+	//var pushBack error
 	// Insert all the headers prioritised by the contained block number
 	inserts := make([]*types.Header, 0, len(headers))
 	for _, header := range headers {
@@ -328,18 +325,44 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
 			log.Warn("Header already scheduled for block fetch", "number", header.Number, "hash", hash)
 			continue
 		}
+		q.blockTaskPool[hash] = header
+		q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
+		// Queue for receipt retrieval
+		if q.mode == FastSync {
 			if _, ok := q.receiptTaskPool[hash]; ok {
 				log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash)
 				continue
 			}
-		// Queue the header for content retrieval
-		q.blockTaskPool[hash] = header
-		q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
-
-		if q.mode == FastSync {
 			q.receiptTaskPool[hash] = header
 			q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
 		}
+		//var bodyNeeded = !header.EmptyBody()
+		//var receiptNeeded = q.mode == FastSync && !header.EmptyReceipts()
+
+		//if pushBack == nil {
+		//	bodyNeeded, receiptNeeded, _, pushBack = q.resultCache.AddFetch(header, q.mode == FastSync)
+		//}
+		//if !receiptNeeded {
+		//	bodyNeeded = true
+		//}
+		// Queue for body retrieval - unless empty block
+		//if bodyNeeded {
+		//	q.blockTaskPool[hash] = header
+		//	q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))
+		//} else { // otherwise, straight to done
+		//	q.blockDonePool[hash] = struct{}{}
+		//}
+		//if receiptNeeded {
+		//	// Queue for receipt retrieval
+		//	if _, ok := q.receiptTaskPool[hash]; ok {
+		//		log.Warn("Header already scheduled for receipt fetch", "number", header.Number, "hash", hash)
+		//		continue
+		//	}
+		//	q.receiptTaskPool[hash] = header
+		//	q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
+		//} else { // done already
+		//	q.receiptDonePool[hash] = struct{}{}
+		//}
 		inserts = append(inserts, header)
 		q.headerHead = hash
 		from++
@@ -350,39 +373,29 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
 // Results retrieves and permanently removes a batch of fetch results from
 // the cache. the result slice will be empty if the queue has been closed.
 func (q *queue) Results(block bool) []*fetchResult {
+
+	// abort early if there are no items and non-blocking requested
+	if !q.resultCache.HasCompletedItems() && !block {
+		return nil
+	}
+
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	// Count the number of items available for processing
+	results := q.resultCache.GetCompleted(maxResultsProcess)
-	nproc := q.countProcessableItems()
+	for len(results) == 0 && !q.closed {
-	for nproc == 0 && !q.closed {
 		if !block {
 			return nil
 		}
 		q.active.Wait()
-		nproc = q.countProcessableItems()
+		results = q.resultCache.GetCompleted(maxResultsProcess)
 	}
-	// Since we have a batch limit, don't pull more into "dangling" memory
+	// Mark results as done
-	if nproc > maxResultsProcess {
-		nproc = maxResultsProcess
-	}
-	results := make([]*fetchResult, nproc)
-	copy(results, q.resultCache[:nproc])
-	if len(results) > 0 {
-		// Mark results as done before dropping them from the cache.
 	for _, result := range results {
 		hash := result.Header.Hash()
 		delete(q.blockDonePool, hash)
 		delete(q.receiptDonePool, hash)
 	}
-		// Delete the results from the cache and clear the tail.
-		copy(q.resultCache, q.resultCache[nproc:])
-		for i := len(q.resultCache) - nproc; i < len(q.resultCache); i++ {
-			q.resultCache[i] = nil
-		}
-		// Advance the expected block number of the first cache entry.
-		q.resultOffset += uint64(nproc)
-
 	// Recalculate the result item weights to prevent memory exhaustion
 	for _, result := range results {
 		size := result.Header.Size()
@@ -397,18 +410,12 @@ func (q *queue) Results(block bool) []*fetchResult {
 		}
 		q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize
 	}
-	}
 	return results
 }
 
 // countProcessableItems counts the processable items.
 func (q *queue) countProcessableItems() int {
-	for i, result := range q.resultCache {
+	return q.resultCache.CountCompleted()
-		if result == nil || result.Pending > 0 {
-			return i
-		}
-	}
-	return len(q.resultCache)
 }
 
 // ReserveHeaders reserves a set of headers for the given peer, skipping any
@@ -455,26 +462,39 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
 // previously failed downloads. Beside the next batch of needed fetches, it also
 // returns a flag whether empty blocks were queued requiring processing.
 func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, error) {
-	isNoop := func(header *types.Header) bool {
-		return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
-	}
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, isNoop)
+	request, bleh, e := q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, 0x01)
+
+	//if request != nil {
+	//	info := []string{}
+	//	for _, x := range request.Headers {
+	//		info = append(info, fmt.Sprintf("%d ", x.Number))
+	//	}
+	//	fmt.Printf("peer %v reserved bodies: %v\n", p.id, info)
+	//}
+	return request, bleh, e
+
 }
 
 // ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
 // any previously failed downloads. Beside the next batch of needed fetches, it
 // also returns a flag whether empty receipts were queued requiring importing.
 func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, error) {
-	isNoop := func(header *types.Header) bool {
-		return header.ReceiptHash == types.EmptyRootHash
-	}
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, isNoop)
+	request, bleh, e := q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, 0x02)
+
+	//if request != nil {
+	//	info := []string{}
+	//	for _, x := range request.Headers {
+	//		info = append(info, fmt.Sprintf("%d ", x.Number))
+	//	}
+	//	fmt.Printf("peer %v reserved receipts: %v\n", p.id, info)
+	//}
+	return request, bleh, e
 }
 
 // reserveHeaders reserves a set of data download operations for a given peer,
@@ -485,7 +505,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo
 // reason the lock is not obtained in here is because the parameters already need
 // to access the queue, so they already need a lock anyway.
 func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
-	pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) {
+	pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, bitFlag uint8) (*fetchRequest, bool, error) {
 	// Short circuit if the pool has been depleted, or if the peer's already
 	// downloading something (sanity check not to corrupt state)
 	if taskQueue.Empty() {
@@ -500,37 +520,36 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 	// Retrieve a batch of tasks, skipping previously failed ones
 	send := make([]*types.Header, 0, count)
 	skip := make([]*types.Header, 0)
-
 	progress := false
+
 	for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
 		header := taskQueue.PopItem().(*types.Header)
 		hash := header.Hash()
-
+		stale, item, _ := q.resultCache.AddFetch(header, q.mode == FastSync)
-		// If we're the first to request this task, initialise the result container
+		if stale {
-		index := int(header.Number.Int64() - int64(q.resultOffset))
+			// Don't put back in the task queue, this item has already been
-		if index >= len(q.resultCache) || index < 0 {
+			// delivered upstream
-			common.Report("index allocation went beyond available resultCache space")
+			progress = true
-			return nil, false, errInvalidChain
-		}
-		if q.resultCache[index] == nil {
-			components := 1
-			if q.mode == FastSync {
-				components = 2
-			}
-			q.resultCache[index] = &fetchResult{
-				Pending: components,
-				Hash:    hash,
-				Header:  header,
-			}
-		}
-		// If this fetch task is a noop, skip this fetch operation
-		if isNoop(header) {
 			donePool[hash] = struct{}{}
 			delete(taskPool, hash)
-
+			proc--
-			space, proc = space-1, proc-1
+			continue
-			q.resultCache[index].Pending--
+		}
+		if item == nil {
+			// There are no resultslots available. Put it back in the task queue
+			taskQueue.Push(header, -int64(header.Number.Uint64()))
+			// Set progress to true -- otherwise the peer will get
+			// penalized for not 'accepting' requests, while in fact it's
+			// because we don't have room for more results right now
 			progress = true
+			break
+		}
+		// Any work to be done?
+		if item.Pending&bitFlag == 0 {
+			progress = true
+			donePool[hash] = struct{}{}
+			delete(taskPool, hash)
+			proc--
 			continue
 		}
 		// Otherwise unless the peer is known not to have the data, add to the retrieve list
@@ -544,7 +563,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 	for _, header := range skip {
 		taskQueue.Push(header, -int64(header.Number.Uint64()))
 	}
-	if progress {
+	if q.resultCache.HasCompletedItems() {
 		// Wake Results, resultCache was modified
 		q.active.Signal()
 	}
@@ -558,7 +577,6 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 		Time:    time.Now(),
 	}
 	pendPool[p.id] = request
-
 	return request, progress, nil
 }
 
@@ -776,11 +794,11 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 // also wakes any threads waiting for data delivery.
 func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) {
 
-	validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
+	validate := func(index int, header *types.Header) error {
-		if types.DeriveSha(types.Transactions(txLists[index])) != txHash {
+		if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash {
 			return errInvalidBody
 		}
-		if types.CalcUncleHash(uncleLists[index]) != uncleHash {
+		if types.CalcUncleHash(uncleLists[index]) != header.UncleHash {
 			return errInvalidBody
 		}
 		return nil
@@ -789,6 +807,8 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi
 	reconstruct := func(index int, result *fetchResult) {
 		result.Transactions = txLists[index]
 		result.Uncles = uncleLists[index]
+		// clear body flag, AND with 1111 1110
+		result.Pending &= 0xfe
 	}
 	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct)
 }
@@ -798,14 +818,16 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi
 // and also wakes any threads waiting for data delivery.
 func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, error) {
 
-	validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
+	validate := func(index int, header *types.Header) error {
-		if types.DeriveSha(types.Receipts(receiptList[index])) != receiptHash {
+		if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash {
 			return errInvalidReceipt
 		}
 		return nil
 	}
 	reconstruct := func(index int, result *fetchResult) {
 		result.Receipts = receiptList[index]
+		// clear bit 1, AND with 1111 1101
+		result.Pending &= 0xfd
 	}
 	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct)
 }
@@ -815,7 +837,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int,
 // This method obtains the lock as needed
 func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
 	pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer,
-	results int, validate func(index int, txHash, uncleHash, receiptHash common.Hash) error, reconstruct func(index int, result *fetchResult)) (int, error) {
+	results int, validate func(index int, header *types.Header) error, reconstruct func(index int, result *fetchResult)) (int, error) {
 
 	q.lock.Lock()
 	// Short circuit if the data was never requested
@@ -840,41 +862,42 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
 		failure error
 		i       int
 	)
-	// Need the read lock to access resultcache
-	q.lock.RLock()
 	for _, header := range request.Headers {
 		// Short circuit assembly if no more fetch results are found
 		if i >= results {
 			break
 		}
 		// Validate the fields
-		if err := validate(i, header.TxHash, header.UncleHash, header.ReceiptHash); err != nil {
+		if err := validate(i, header); err != nil {
 			failure = err
 			break
 		}
 		header.Hash()
 		i++
 	}
-	q.lock.RUnlock()
 	q.lock.Lock()
 	var acceptCount = 0
 	for _, header := range request.Headers[:i] {
-		index := int(header.Number.Int64() - int64(q.resultOffset))
+		// TODO @holiman
-		if index >= len(q.resultCache) || index < 0 {
+		// q.resultCache.deliver(header, data, reconstruct)
-			// TODO! this should probably be errStaleDelivery instead
+		// or
-			failure = errStaleDelivery
+		// q.resultCache.deliverBody , q.resultCache.deliverReceipts
-			break
+		// or
-		}
+		// q.resultCache.deliverBodies( bodies, headers[0])
-		if res := q.resultCache[index]; res != nil {
+
-			hash := header.Hash()
+		if res, stale, err := q.resultCache.GetFetchResult(header); err == nil {
-			donePool[hash] = struct{}{}
 			reconstruct(acceptCount, res)
-			res.Pending--
+		} else {
-			delete(taskPool, hash)
-		}
 			// else: betweeen here and above, some other peer filled this result
 			// we just ignore and move on
-
+			// TODO @holiman
+			// figure out how this can happen (it shouldn't)
+			log.Info("delivery stale?", "err", err, "stale", stale)
+			failure = errStaleDelivery
+		}
+		hash := header.Hash()
+		donePool[hash] = struct{}{}
+		delete(taskPool, hash)
 		// Clean up a successful fetch
 		request.Headers[acceptCount] = nil
 		acceptCount++
@@ -909,8 +932,6 @@ func (q *queue) Prepare(offset uint64, mode SyncMode) {
 	defer q.lock.Unlock()
 
 	// Prepare the queue for sync results
-	if q.resultOffset < offset {
+	q.resultCache.Prepare(offset)
-		q.resultOffset = offset
-	}
 	q.mode = mode
 }

eth/downloader/resultcache.go

@@ -0,0 +1,258 @@
+// Copyright 2019 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// resultcache implements a structure for maintaining fetchResults, tracking their
+// download-progress and delivering (finished) results
+
+package downloader
+
+import (
+	"fmt"
+	"github.com/ethereum/go-ethereum/log"
+	"sync"
+	"sync/atomic"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type resultStore struct {
+	items        []*fetchResult     // Downloaded but not yet delivered fetch results
+	lock         *sync.RWMutex      // lock protect internals
+	resultOffset uint64             // Offset of the first cached fetch result in the block chain
+	resultSize   common.StorageSize // Approximate size of a block (exponential moving average)
+
+	// Internal index of first non-completed entry, updated atomically when needed.
+	// If all items are complete, this will equal length(items), so
+	// *important* : is not safe to use for indexing without checking against length
+	indexIncomplete int32
+}
+
+func newResultStore(size int) *resultStore {
+	return &resultStore{
+		resultOffset: 0,
+		items:        make([]*fetchResult, size),
+		resultSize:   0, // TODO: use a saner default, left at zero as it was legacy
+		lock:         new(sync.RWMutex),
+	}
+}
+
+// AddFetch adds a header for body/receipt fetching.
+// returning
+// stale -- if true, this item is already passed, and should not be requested again
+// fetchResult -- if `nil`, that means no fetch was created, and that the
+// if an error is returned, that most likely means backpressure prevents the results from expanding,
+// and someone needs to take care of results
+func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale bool, item *fetchResult, err error) {
+	hash := header.Hash()
+	r.lock.RLock()
+	item, _, stale, err = r.getFetchResult(header)
+	if err != nil {
+		r.lock.RUnlock()
+		log.Info("resultcache addfetch err [1]", "error", err.Error())
+		// can't create a fetchResult right away
+		return stale, nil, err
+	}
+	if item != nil {
+		r.lock.RUnlock()
+		return false, item, nil
+	}
+	r.lock.RUnlock()
+	// Need to create a fetchresult, and as we've just release the Rlock,
+	// we need to check again after obtaining the writelock
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	var index int
+	item, index, stale, err = r.getFetchResult(header)
+	if err != nil {
+		// can't create a fetchResult right away
+		log.Info("resultcache addfetch err [2]", "error", err.Error())
+		return stale, nil, err
+
+	}
+	if item == nil {
+		item = &fetchResult{
+			Hash:   hash,
+			Header: header,
+		}
+		// Need to fetch body?
+		if !header.EmptyBody() {
+			// yes
+			item.Pending |= 0x1
+		}
+		// Do we need to fetch receipts?
+		if fastSync && !header.EmptyReceipts() {
+			item.Pending |= 0x2
+		}
+		r.items[index] = item
+	}
+	return false, item, nil
+
+}
+
+func (r *resultStore) GetFetchResult(header *types.Header) (*fetchResult, bool, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	res, _, stale, err := r.getFetchResult(header)
+	return res, stale, err
+}
+
+// getFetchResult returns the fetchResult corresponding to the given item, and the index where
+// the result is stored.
+// There are two ways it can error:
+// 1. The header is too far off in the future, and we don't have room for it.
+// 2. The header is stale, and the results for that header has already been delivered upstream.
+func (r *resultStore) getFetchResult(header *types.Header) (item *fetchResult, index int, stale bool, err error) {
+	index = int(header.Number.Int64() - int64(r.resultOffset))
+	if index >= len(r.items) {
+		err = fmt.Errorf("index allocation went beyond available resultStore space "+
+			"(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d",
+			index, header.Number.Int64(), r.resultOffset, len(r.items))
+		return item, index, stale, err
+	}
+	if index < 0 {
+		stale = true
+		err = fmt.Errorf("index allocation went beyond available resultStore space "+
+			"(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d",
+			index, header.Number.Int64(), r.resultOffset, len(r.items))
+		return item, index, stale, err
+	}
+	item = r.items[index]
+	return item, index, stale, err
+}
+
+// numberSpan returns the header number start and end, for the headers
+// currently "allocated" for download (both completed, in-flight and pending)
+func (r *resultStore) NumberSpan() (uint64, uint64) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	return r.resultOffset, r.resultOffset + uint64(len(r.items))
+
+}
+
+// hasCompletedItems returns true if there are processable items available
+// this method is cheaper than countCompleted
+func (r *resultStore) HasCompletedItems() bool {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	if len(r.items) == 0 {
+		return false
+	}
+	if item := r.items[0]; item != nil && item.Pending == 0 {
+		return true
+	}
+	return false
+}
+
+// CountCompleted returns the number of items completed
+func (r *resultStore) CountCompleted() int {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	return r.countCompleted()
+}
+
+// countCompleted returns the number of items completed
+// assumes (at least) rlock is held
+func (r *resultStore) countCompleted() int {
+	// We iterate from the already known complete point, and see
+	// if any more has completed since last count
+	// debug
+	/*
+		var (
+			nils         = 0
+			fins         = 0
+			bodyneeds    = 0
+			receiptneeds = 0
+		)
+		var ctx []interface{}
+		for _, item := range r.items {
+			if item == nil {
+				nils++
+			} else {
+				if item.Pending == 0 {
+					fins++
+				} else {
+					if item.Pending&0x01 != 0 {
+						bodyneeds++
+					} else {
+						receiptneeds++
+					}
+				}
+			}
+		}
+		ctx = append(ctx, "items", len(r.items), "nils", nil, "fins", fins,
+			"needB", bodyneeds, "needR", receiptneeds)
+	*/
+	/// end debug
+	index := atomic.LoadInt32(&r.indexIncomplete)
+	for ; ; index++ {
+		if index >= int32(len(r.items)) {
+			break
+		}
+		result := r.items[index]
+		if result == nil || result.Pending > 0 {
+			break
+		}
+	}
+	/*
+		if index < int32(len(r.items)) {
+			//ctx = append(ctx, []interface{}{"index", index, "blocknum", uint64(index) + r.resultOffset}...)
+			if r.items[index] != nil {
+				log.Info("resultstore", ctx...)
+			} else {
+				ctx = append(ctx, []interface{}{"first missing", "nil"}...)
+				log.Info("resultstore", ctx...)
+			}
+		} else {
+			ctx = append(ctx, []interface{}{"first missing", "out of range"}...)
+			log.Info("resultstore", ctx...)
+		}
+	*/
+	atomic.StoreInt32(&r.indexIncomplete, index)
+	return int(index)
+}
+
+// getCompleted returns the next batch of completed fetchresults
+func (r *resultStore) GetCompleted(limit int) []*fetchResult {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	completed := r.countCompleted()
+	if limit > completed {
+		limit = completed
+	}
+	results := make([]*fetchResult, limit)
+	copy(results, r.items[:limit])
+
+	// 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.items[i] = nil
+	}
+	// Advance the expected block number of the first cache entry.
+	r.resultOffset += uint64(limit)
+	// And subtract the number of items from our two indexes
+	atomic.StoreInt32(&r.indexIncomplete, int32(completed-limit))
+	return results
+}
+
+func (r *resultStore) Prepare(offset uint64) {
+	r.lock.Lock()
+	if r.resultOffset < offset {
+		r.resultOffset = offset
+	}
+	r.lock.Unlock()
+}

eth/fetcher/block_fetcher.go

@@ -538,31 +538,42 @@ func (f *BlockFetcher) loop() {
 				return
 			}
 			bodyFilterInMeter.Mark(int64(len(task.transactions)))
-
 			blocks := []*types.Block{}
+			// abort early if there's nothing explicitly requested
+			if len(f.completing) > 0 {
 				for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ {
 					// Match up a body to any possible completion request
-				matched := false
+					var (
-
+						matched   = false
+						uncleHash common.Hash // calculated lazily and reused
+						txnHash   common.Hash // calculated lazily and reused
+					)
 					for hash, announce := range f.completing {
-					if f.queued[hash] == nil {
+						if f.queued[hash] != nil || announce.origin != task.peer {
-						txnHash := types.DeriveSha(types.Transactions(task.transactions[i]))
+							continue
-						uncleHash := types.CalcUncleHash(task.uncles[i])
+						}
-
+						if uncleHash == (common.Hash{}) {
-						if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash && announce.origin == task.peer {
+							uncleHash = types.CalcUncleHash(task.uncles[i])
+						}
+						if uncleHash != announce.header.UncleHash {
+							continue
+						}
+						if txnHash == (common.Hash{}) {
+							txnHash = types.DeriveSha(types.Transactions(task.transactions[i]))
+						}
+						if txnHash != announce.header.TxHash {
+							continue
+						}
 						// Mark the body matched, reassemble if still unknown
 						matched = true
-
 						if f.getBlock(hash) == nil {
 							block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])
 							block.ReceivedAt = task.time
-
 							blocks = append(blocks, block)
 						} else {
 							f.forgetHash(hash)
 						}
-						}
+
-					}
 					}
 					if matched {
 						task.transactions = append(task.transactions[:i], task.transactions[i+1:]...)
@@ -571,7 +582,7 @@ func (f *BlockFetcher) loop() {
 						continue
 					}
 				}
-
+			}
 			bodyFilterOutMeter.Mark(int64(len(task.transactions)))
 			select {
 			case filter <- task: