downloader/queue: increase underlying buffer of results, new throttle mechanism

eth/downloader/downloader.go

@@ -1091,9 +1091,8 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) (
 			return d.queue.DeliverHeaders(pack.peerID, pack.headers, d.headerProcCh)
 		}
 		expire  = func() map[string]int { return d.queue.ExpireHeaders(d.requestTTL()) }
-		throttle = func() bool { return false }
+		reserve = func(p *peerConnection, count int) (*fetchRequest, bool, bool, error) {
-		reserve  = func(p *peerConnection, count int) (*fetchRequest, bool, error) {
+			return d.queue.ReserveHeaders(p, count), false, false, nil
-			return d.queue.ReserveHeaders(p, count), false, nil
 		}
 		fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchHeaders(req.From, MaxHeaderFetch) }
 		capacity = func(p *peerConnection) int { return p.HeaderCapacity(d.requestRTT()) }
@@ -1102,7 +1101,7 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) (
 		}
 	)
 	err := d.fetchParts(d.headerCh, deliver, d.queue.headerContCh, expire,
-		d.queue.PendingHeaders, d.queue.InFlightHeaders, throttle, reserve,
+		d.queue.PendingHeaders, d.queue.InFlightHeaders, reserve,
 		nil, fetch, d.queue.CancelHeaders, capacity, d.peers.HeaderIdlePeers, setIdle, "headers")
 
 	log.Debug("Skeleton fill terminated", "err", err)
@@ -1128,7 +1127,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
 		setIdle  = func(p *peerConnection, accepted int, deliveryTime time.Time) { p.SetBodiesIdle(accepted, deliveryTime) }
 	)
 	err := d.fetchParts(d.bodyCh, deliver, d.bodyWakeCh, expire,
-		d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ShouldThrottleBlocks, d.queue.ReserveBodies,
+		d.queue.PendingBlocks, d.queue.InFlightBlocks, d.queue.ReserveBodies,
 		d.bodyFetchHook, fetch, d.queue.CancelBodies, capacity, d.peers.BodyIdlePeers, setIdle, "bodies")
 
 	log.Debug("Block body download terminated", "err", err)
@@ -1154,7 +1153,7 @@ func (d *Downloader) fetchReceipts(from uint64) error {
 		}
 	)
 	err := d.fetchParts(d.receiptCh, deliver, d.receiptWakeCh, expire,
-		d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ShouldThrottleReceipts, d.queue.ReserveReceipts,
+		d.queue.PendingReceipts, d.queue.InFlightReceipts, d.queue.ReserveReceipts,
 		d.receiptFetchHook, fetch, d.queue.CancelReceipts, capacity, d.peers.ReceiptIdlePeers, setIdle, "receipts")
 
 	log.Debug("Transaction receipt download terminated", "err", err)
@@ -1187,7 +1186,7 @@ func (d *Downloader) fetchReceipts(from uint64) error {
 //  - setIdle:     network callback to set a peer back to idle and update its estimated capacity (traffic shaping)
 //  - kind:        textual label of the type being downloaded to display in log mesages
 func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack) (int, error), wakeCh chan bool,
-	expire func() map[string]int, pending func() int, inFlight func() bool, throttle func() bool, reserve func(*peerConnection, int) (*fetchRequest, bool, error),
+	expire func() map[string]int, pending func() int, inFlight func() bool, reserve func(*peerConnection, int) (*fetchRequest, bool, bool, error),
 	fetchHook func([]*types.Header), fetch func(*peerConnection, *fetchRequest) error, cancel func(*fetchRequest), capacity func(*peerConnection) int,
 	idle func() ([]*peerConnection, int), setIdle func(*peerConnection, int, time.Time), kind string) error {
 
@@ -1306,34 +1305,32 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 			// Send a download request to all idle peers, until throttled
 			progressed, throttled, running := false, false, inFlight()
 			idles, total := idle()
-
+			pendCount := 1
 			for _, peer := range idles {
 				// Short circuit if throttling activated
-				if throttle() {
+				if throttled {
-					throttled = true
 					break
 				}
 				// Short circuit if there is no more available task.
-				if pending() == 0 {
+				if pendCount = pending(); pendCount == 0 {
 					break
 				}
 				// Reserve a chunk of fetches for a peer. A nil can mean either that
 				// no more headers are available, or that the peer is known not to
 				// have them.
-				request, progress, err := reserve(peer, capacity(peer))
+				request, progress, throttle, err := reserve(peer, capacity(peer))
 				if err != nil {
+					log.Info("Error in loop", "err", err)
 					return err
 				}
 				if progress {
 					progressed = true
 				}
-				if request == nil {
+				if throttle {
-					// This means we've over-committed, and reserving returns nil
-					// because there's no space for the top-prio header to download.
-					// No need to continue this loop
 					throttled = true
-					break
+					throttleBlockCounter.Inc(1)
 				}
+				if request != nil {
 					if request.From > 0 {
 						peer.log.Trace("Requesting new batch of data", "type", kind, "from", request.From)
 					} else {
@@ -1351,11 +1348,12 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 						// a much bigger issue.
 						panic(fmt.Sprintf("%v: %s fetch assignment failed", peer, kind))
 					}
+				}
 				running = true
 			}
 			// Make sure that we have peers available for fetching. If all peers have been tried
 			// and all failed throw an error
-			if !progressed && !throttled && !running && len(idles) == total && pending() > 0 {
+			if !progressed && !throttled && !running && len(idles) == total && pendCount > 0 {
 				return errPeersUnavailable
 			}
 		}
@@ -1482,7 +1480,7 @@ func (d *Downloader) processHeaders(origin uint64, pivot uint64, td *big.Int) er
 							rollbackErr = err
 							rollback = append(rollback, chunk[:n]...)
 						}
-						log.Info("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, "parent", chunk[n].ParentHash)
 						return errInvalidChain
 					}
 					// All verifications passed, store newly found uncertain headers
@@ -1637,6 +1635,7 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error {
 			}
 		}
 		P, beforeP, afterP := splitAroundPivot(pivot, results)
+		results = nil
 		if err := d.commitFastSyncData(beforeP, sync); err != nil {
 			return err
 		}
@@ -1674,6 +1673,15 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error {
 }
 
 func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) {
+	if len(results) == 0 {
+		return
+	}
+	if lastNum := results[len(results)-1].Header.Number.Uint64(); lastNum < pivot {
+		// the pivot is somewhere in the future
+		before = results
+		return
+	}
+	// This can also be optimized, but only happens very seldom
 	for _, result := range results {
 		num := result.Header.Number.Uint64()
 		switch {
@@ -1704,7 +1712,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.Info("Inserting fast-sync blocks", "items", len(results),
+	log.Debug("Inserting fast-sync blocks", "items", len(results),
 		"firstnum", first.Number, "firsthash", first.Hash(),
 		"lastnumn", last.Number, "lasthash", last.Hash(),
 	)

eth/downloader/queue.go

@@ -23,6 +23,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -32,6 +33,11 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 )
 
+const (
+	BodyType    = 0
+	ReceiptType = 1
+)
+
 var (
 	blockCacheItems      = 8192             // Maximum number of blocks to cache before throttling the download
 	blockCacheMemory     = 64 * 1024 * 1024 // Maximum amount of memory to use for block caching
@@ -54,7 +60,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 uint8       // Number of data fetches still pending
+	pending int32       // Flag telling what deliveries are outstanding
 	Hash    common.Hash // Hash of the header to prevent recalculating
 
 	Header       *types.Header
@@ -63,6 +69,52 @@ type fetchResult struct {
 	Receipts     types.Receipts
 }
 
+func newFetchResult(header *types.Header, fastSync bool) *fetchResult {
+	item := &fetchResult{
+		Hash:   header.Hash(),
+		Header: header,
+	}
+	if !header.EmptyBody() {
+		item.pending = 1
+	}
+	if fastSync && !header.EmptyReceipts() {
+		item.pending += 2
+	}
+	return item
+}
+
+// SetBodyDone flags the body as finished.
+func (f *fetchResult) SetBodyDone() {
+	if v := atomic.LoadInt32(&f.pending); v == 1 || v == 3 {
+		atomic.AddInt32(&f.pending, -1)
+	}
+}
+
+// AllDone checks item is done
+func (f *fetchResult) AllDone() bool {
+	return atomic.LoadInt32(&f.pending) == 0
+}
+
+// SetReceiptsDone flags the receipts as finished.
+func (f *fetchResult) SetReceiptsDone() {
+	if v := atomic.LoadInt32(&f.pending); v == 2 || v == 3 {
+		atomic.AddInt32(&f.pending, -2)
+	}
+}
+
+// CheckDone checks if the given type is done already
+func (f *fetchResult) Done(typ int) bool {
+	v := atomic.LoadInt32(&f.pending)
+	switch typ {
+	case BodyType:
+		return !(v == 1 || v == 3)
+	case ReceiptType:
+		return !(v == 2 || v == 3)
+	default:
+		return false
+	}
+}
+
 // queue represents hashes that are either need fetching or are being fetched
 type queue struct {
 	mode SyncMode // Synchronisation mode to decide on the block parts to schedule for fetching
@@ -82,12 +134,10 @@ type queue struct {
 	blockTaskPool  map[common.Hash]*types.Header // [eth/62] Pending block (body) retrieval tasks, mapping hashes to headers
 	blockTaskQueue *prque.Prque                  // [eth/62] Priority queue of the headers to fetch the blocks (bodies) for
 	blockPendPool  map[string]*fetchRequest      // [eth/62] Currently pending block (body) retrieval operations
-	blockDonePool  map[common.Hash]struct{}      // [eth/62] Set of the completed block (body) fetches
 
 	receiptTaskPool  map[common.Hash]*types.Header // [eth/63] Pending receipt retrieval tasks, mapping hashes to headers
 	receiptTaskQueue *prque.Prque                  // [eth/63] Priority queue of the headers to fetch the receipts for
 	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 *resultStore // Downloaded but not yet delivered fetch results
 	//resultOffset uint64             // Offset of the first cached fetch result in the block chain
@@ -107,13 +157,10 @@ func newQueue(blockCacheLimit int) *queue {
 		blockTaskPool:    make(map[common.Hash]*types.Header),
 		blockTaskQueue:   prque.New(nil),
 		blockPendPool:    make(map[string]*fetchRequest),
-		blockDonePool:    make(map[common.Hash]struct{}),
 		receiptTaskPool:  make(map[common.Hash]*types.Header),
 		receiptTaskQueue: prque.New(nil),
 		receiptPendPool:  make(map[string]*fetchRequest),
-		receiptDonePool:  make(map[common.Hash]struct{}),
+		resultCache:      newResultStore(blockCacheLimit * 2),
-		//resultCache:      make([]*fetchResult, blockCacheItems),
-		resultCache: newResultStore(blockCacheLimit),
 		active:           sync.NewCond(lock),
 		lock:             lock,
 	}
@@ -133,14 +180,12 @@ func (q *queue) Reset() {
 	q.blockTaskPool = make(map[common.Hash]*types.Header)
 	q.blockTaskQueue.Reset()
 	q.blockPendPool = make(map[string]*fetchRequest)
-	q.blockDonePool = make(map[common.Hash]struct{})
 
 	q.receiptTaskPool = make(map[common.Hash]*types.Header)
 	q.receiptTaskQueue.Reset()
 	q.receiptPendPool = make(map[string]*fetchRequest)
-	q.receiptDonePool = make(map[common.Hash]struct{})
 
-	q.resultCache = newResultStore(blockCacheItems)
+	q.resultCache = newResultStore(blockCacheItems * 2)
 }
 
 // Close marks the end of the sync, unblocking Results.
@@ -210,54 +255,8 @@ func (q *queue) Idle() bool {
 
 	queued := q.blockTaskQueue.Size() + q.receiptTaskQueue.Size()
 	pending := len(q.blockPendPool) + len(q.receiptPendPool)
-	cached := len(q.blockDonePool) + len(q.receiptDonePool)
 
-	return (queued + pending + cached) == 0
+	return (queued + pending) == 0
-}
-
-// ShouldThrottleBlocks checks if the download should be throttled (active block (body)
-// fetches exceed block cache).
-func (q *queue) ShouldThrottleBlocks() bool {
-	q.lock.Lock()
-	t := q.resultSlots(q.blockPendPool, q.blockDonePool) <= 0
-	q.lock.Unlock()
-	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()
-	t := q.resultSlots(q.receiptPendPool, q.receiptDonePool) <= 0
-	q.lock.Unlock()
-	if t {
-		throttleReceiptCounter.Inc(1)
-	}
-	return t
-}
-
-// resultSlots calculates the number of results slots available for requests
-// whilst adhering to both the item and the memory limit too of the results
-// cache.
-func (q *queue) resultSlots(pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}) int {
-	// Calculate the maximum length capped by the memory limit
-	cacheItems := len(q.resultCache.items)
-	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 := q.resultCache.countCompleted()
-	// Calculate the number of slots currently downloading
-	pending := 0
-	//iterations := 0
-	for _, request := range pendPool {
-		pending += len(request.Headers)
-	}
-	return limit - finished - pending
 }
 
 // ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill
@@ -372,32 +371,26 @@ 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.
+// This is 'thread-safe', but assumes that there are not two simultaneous
+// callers to Results (both will modify q.resultSize)
 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()
-
 	results := q.resultCache.GetCompleted(maxResultsProcess)
 	for len(results) == 0 && !q.closed {
 		if !block {
 			return nil
 		}
+		q.lock.Lock()
 		q.active.Wait()
+		q.lock.Unlock()
 		results = q.resultCache.GetCompleted(maxResultsProcess)
 	}
-	// Mark results as done
 	for _, result := range results {
-		hash := result.Header.Hash()
-		delete(q.blockDonePool, hash)
-		delete(q.receiptDonePool, hash)
-	}
 		// Recalculate the result item weights to prevent memory exhaustion
-	for _, result := range results {
 		size := result.Header.Size()
 		for _, uncle := range result.Uncles {
 			size += uncle.Size()
@@ -408,14 +401,33 @@ func (q *queue) Results(block bool) []*fetchResult {
 		for _, tx := range result.Transactions {
 			size += tx.Size()
 		}
-		q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize
+		q.resultSize = common.StorageSize(blockCacheSizeWeight)*size +
+			(1-common.StorageSize(blockCacheSizeWeight))*q.resultSize
+	}
+	// Using the newly calibrated resultsize, figure out the new throttle limit
+	// on the result cache
+	throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize)
+	q.resultCache.SetThrottleThreshold(throttleThreshold)
+	// log some info at certain times
+	if time.Now().Second()&0xa == 0 {
+		info := q.Stats()
+		info = append(info, "throttle", throttleThreshold)
+		log.Info("queue stats", info...)
 	}
 	return results
 }
 
-// countProcessableItems counts the processable items.
+func (q *queue) Stats() []interface{} {
-func (q *queue) countProcessableItems() int {
+	q.lock.RLock()
-	return q.resultCache.CountCompleted()
+	defer q.lock.RUnlock()
+	return q.stats()
+}
+func (q *queue) stats() []interface{} {
+	return []interface{}{
+		"receiptTaskQueue", q.receiptTaskQueue.Size(),
+		"blockTaskQueue", q.blockTaskQueue.Size(),
+		"est resultSize", q.resultSize,
+	}
 }
 
 // ReserveHeaders reserves a set of headers for the given peer, skipping any
@@ -461,40 +473,21 @@ func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
 // ReserveBodies reserves a set of body fetches for the given peer, skipping any
 // 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) {
+func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, bool, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	request, bleh, e := q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, 0x01)
+	return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, BodyType)
-
-	//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) {
+func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, bool, error) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	request, bleh, e := q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, 0x02)
+	return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, ReceiptType)
-
-	//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,
@@ -504,52 +497,68 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo
 // Note, this method expects the queue lock to be already held for writing. The
 // 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.
+// returns:
+// item - the fetchRequest
+// progress, bool - whether any progress was made
+// throttle, bool - if the caller should throttle for a while
+// error - any error that occcurred
 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{}, bitFlag uint8) (*fetchRequest, bool, error) {
+	pendPool map[string]*fetchRequest, typ int) (*fetchRequest, bool, 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() {
-		return nil, false, nil
+		return nil, false, true, nil
 	}
 	if _, ok := pendPool[p.id]; ok {
-		return nil, false, nil
+		return nil, false, false, nil
 	}
-	// Calculate an upper limit on the items we might fetch (i.e. throttling)
-	space := q.resultSlots(pendPool, donePool)
-
 	// Retrieve a batch of tasks, skipping previously failed ones
 	send := make([]*types.Header, 0, count)
 	skip := make([]*types.Header, 0)
 	progress := false
-
+	throttled := false
-	for proc := 0; proc < space && len(send) < count && !taskQueue.Empty(); proc++ {
+	for proc := 0; len(send) < count && !taskQueue.Empty(); proc++ {
+		// the task queue will pop items in order, so the highest prio block
+		// is also the lowest block number.
 		header := taskQueue.PopItem().(*types.Header)
+		// we can ask the resultcache if this header is within the
+		// "prioritized" segment of blocks. If it is not, we need to throttle
+
 		hash := header.Hash()
-		stale, item, _ := q.resultCache.AddFetch(header, q.mode == FastSync)
+		stale, throttle, item, err := q.resultCache.AddFetch(header, q.mode == FastSync)
 		if stale {
 			// Don't put back in the task queue, this item has already been
 			// delivered upstream
 			progress = true
-			donePool[hash] = struct{}{}
 			delete(taskPool, hash)
-			proc--
+			proc = proc - 1
 			continue
 		}
-		if item == nil {
+		if throttle {
 			// 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
+			// However, if there are any left as 'skipped', we should not tell
+			// the caller to throttle, since we still want some other
+			// peer to fetch those for us
+			throttled = len(skip) == 0
 			break
 		}
-		// Any work to be done?
+		if err != nil {
-		if item.Pending&bitFlag == 0 {
+			// this most definitely should _not_ happen
-			progress = true
+			log.Warn("reserve headers error", "error", err)
-			donePool[hash] = struct{}{}
+			// There are no resultslots available. Put it back in the task queue
+			taskQueue.Push(header, -int64(header.Number.Uint64()))
+			break
+		}
+		if item.Done(typ) {
+			// If it's a noop, we can skip this task
 			delete(taskPool, hash)
-			proc--
+			proc = proc - 1
+			progress = true
 			continue
 		}
 		// Otherwise unless the peer is known not to have the data, add to the retrieve list
@@ -569,7 +578,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 	}
 	// Assemble and return the block download request
 	if len(send) == 0 {
-		return nil, progress, nil
+		return nil, progress, throttled, nil
 	}
 	request := &fetchRequest{
 		Peer:    p,
@@ -577,7 +586,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 		Time:    time.Now(),
 	}
 	pendPool[p.id] = request
-	return request, progress, nil
+	return request, progress, throttled, nil
 }
 
 // CancelHeaders aborts a fetch request, returning all pending skeleton indexes to the queue.
@@ -807,10 +816,10 @@ 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.SetBodyDone()
-		result.Pending &= 0xfe
 	}
-	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct)
+	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool,
+		bodyReqTimer, len(txLists), validate, reconstruct)
 }
 
 // DeliverReceipts injects a receipt retrieval response into the results queue.
@@ -826,18 +835,19 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int,
 	}
 	reconstruct := func(index int, result *fetchResult) {
 		result.Receipts = receiptList[index]
-		// clear bit 1, AND with 1111 1101
+		result.SetReceiptsDone()
-		result.Pending &= 0xfd
 	}
-	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct)
+	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool,
+		receiptReqTimer, len(receiptList), validate, reconstruct)
 }
 
 // deliver injects a data retrieval response into the results queue.
 //
 // This method obtains the lock as needed
-func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
+func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
-	pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer,
+	taskQueue *prque.Prque, pendPool map[string]*fetchRequest, reqTimer metrics.Timer,
-	results int, validate func(index int, header *types.Header) 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
@@ -867,37 +877,27 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
 		if i >= results {
 			break
 		}
+		header.Hash()
 		// Validate the fields
 		if err := validate(i, header); err != nil {
 			failure = err
 			break
 		}
-		header.Hash()
 		i++
 	}
 	q.lock.Lock()
 	var acceptCount = 0
 	for _, header := range request.Headers[:i] {
-		// TODO @holiman
-		// q.resultCache.deliver(header, data, reconstruct)
-		// or
-		// q.resultCache.deliverBody , q.resultCache.deliverReceipts
-		// or
-		// q.resultCache.deliverBodies( bodies, headers[0])
-
 		if res, stale, err := q.resultCache.GetFetchResult(header); err == nil {
 			reconstruct(acceptCount, res)
 		} else {
-			// else: betweeen here and above, some other peer filled this result
+			// else: betweeen here and above, some other peer filled this result,
-			// we just ignore and move on
+			// or it was indeed a no-op. This should not happen, but if it does it's
-			// TODO @holiman
+			// not something to panic about
-			// figure out how this can happen (it shouldn't)
 			log.Info("delivery stale?", "err", err, "stale", stale)
 			failure = errStaleDelivery
 		}
-		hash := header.Hash()
+		delete(taskPool, header.Hash())
-		donePool[hash] = struct{}{}
-		delete(taskPool, hash)
 		// Clean up a successful fetch
 		request.Headers[acceptCount] = nil
 		acceptCount++

eth/downloader/resultcache.go

@@ -21,126 +21,129 @@ 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"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 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
+	indexIncomplete int32 // atomic access
+
+	// 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
 }
 
 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),
+		throttleThreshold: 3 * uint64(size) / 4, // 75%
 	}
 }
 
-// AddFetch adds a header for body/receipt fetching.
+func (r *resultStore) SetThrottleThreshold(threshold uint64) {
-// returning
+	r.lock.Lock()
-// stale -- if true, this item is already passed, and should not be requested again
+	defer r.lock.Unlock()
-// fetchResult -- if `nil`, that means no fetch was created, and that the
+	limit := uint64(len(r.items)) * 3 / 4
-// if an error is returned, that most likely means backpressure prevents the results from expanding,
+	if threshold >= limit {
-// and someone needs to take care of results
+		threshold = limit
-func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale bool, item *fetchResult, err error) {
+	}
-	hash := header.Hash()
+	r.throttleThreshold = threshold
+}
+
+// AddFetch adds a header for body/receipt fetching. This is used when the queue
+// wants to reserve headers for fetching.
+// It returns the following:
+// stale       -- if true, this item is already passed, and should not be requested again.
+// throttled   -- if true, the resultcache is at capacity, and this particular header is not
+//                prio right now
+// fetchResult -- the result to store data into
+// err         -- any error that occurred
+func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) {
+	header.Hash()
 	r.lock.RLock()
-	item, _, stale, err = r.getFetchResult(header)
+	var index int
-	if err != nil {
+	if item, index, stale, throttled, err = r.getFetchResult(header); err != nil {
 		r.lock.RUnlock()
-		log.Info("resultcache addfetch err [1]", "error", err.Error())
+		return
-		// can't create a fetchResult right away
+	}
-		return stale, nil, err
+	if stale {
+		r.lock.RUnlock()
+		return
+	}
+	if throttled {
+		// Index is above the current threshold of 'prioritized' blocks,
+		log.Debug("resultcache throttle", "index", index, "threshold", r.throttleThreshold)
+		r.lock.RUnlock()
+		return
 	}
 	if item != nil {
+		// All good, item already exists (perhaps a receipt fetch following
+		// a body fetch)
 		r.lock.RUnlock()
-		return false, item, nil
+		return
 	}
 	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
+	// Same checks as above, now with wlock
-	item, index, stale, err = r.getFetchResult(header)
+	if item, index, stale, throttled, err = r.getFetchResult(header); err != nil {
-	if err != nil {
+		return
-		// can't create a fetchResult right away
+	}
-		log.Info("resultcache addfetch err [2]", "error", err.Error())
+	if stale || throttled {
-		return stale, nil, err
+		return
-
 	}
 	if item == nil {
-		item = &fetchResult{
+		item = newFetchResult(header, fastSync)
-			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
+	return
-
 }
 
+// GetFetchResult returns the fetchResult for the given header. If the 'stale' flag
+// is true, that means the header has already been delivered 'upstream'.
 func (r *resultStore) GetFetchResult(header *types.Header) (*fetchResult, bool, error) {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
-	res, _, stale, err := r.getFetchResult(header)
+	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:
+func (r *resultStore) getFetchResult(header *types.Header) (item *fetchResult, index int, stale, throttle bool, err 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))
+	throttle = index >= int(r.throttleThreshold)
+	stale = index < 0
+
 	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
+		return
 	}
-	if index < 0 {
+	if stale {
-		stale = true
+		return
-		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
+	return
-}
-
-// 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
@@ -151,81 +154,33 @@ func (r *resultStore) HasCompletedItems() bool {
 	if len(r.items) == 0 {
 		return false
 	}
-	if item := r.items[0]; item != nil && item.Pending == 0 {
+	if item := r.items[0]; item != nil && item.AllDone() {
 		return true
 	}
 	return false
 }
 
-// CountCompleted returns the number of items completed
+// countCompleted returns the number of items ready for delivery, stopping at
-func (r *resultStore) CountCompleted() int {
+// the first non-complete item.
-	r.lock.RLock()
+// It assumes (at least) rlock is held
-	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 {
+		if result == nil || !result.AllDone() {
 			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
+// GetCompleted returns the next batch of completed fetchResults
 func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	r.lock.Lock()
 	defer r.lock.Unlock()
@@ -244,11 +199,12 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	}
 	// Advance the expected block number of the first cache entry.
 	r.resultOffset += uint64(limit)
-	// And subtract the number of items from our two indexes
+	// And subtract the number of items from our index
-	atomic.StoreInt32(&r.indexIncomplete, int32(completed-limit))
+	atomic.AddInt32(&r.indexIncomplete, int32(-limit))
 	return results
 }
 
+// Prepare initialises the offset with the given block number
 func (r *resultStore) Prepare(offset uint64) {
 	r.lock.Lock()
 	if r.resultOffset < offset {