eth/downloader: review concerns + simplify resultcache and queue

eth/downloader/downloader.go

@@ -367,7 +367,7 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode
 		d.stateBloom.Close()
 	}
 	// Reset the queue, peer set and wake channels to clean any internal leftover state
-	d.queue.Reset()
+	d.queue.Reset(blockCacheItems)
 	d.peers.Reset()
 
 	for _, ch := range []chan bool{d.bodyWakeCh, d.receiptWakeCh} {
@@ -1644,7 +1644,6 @@ 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
 		}

eth/downloader/queue.go

@@ -83,7 +83,7 @@ func newFetchResult(header *types.Header, fastSync bool) *fetchResult {
 
 // SetBodyDone flags the body as finished.
 func (f *fetchResult) SetBodyDone() {
-	if v := atomic.LoadInt32(&f.pending); v == 1 || v == 3 {
+	if v := atomic.LoadInt32(&f.pending); (v & 1) != 0 {
 		atomic.AddInt32(&f.pending, -1)
 	}
 }
@@ -95,19 +95,19 @@ func (f *fetchResult) AllDone() bool {
 
 // SetReceiptsDone flags the receipts as finished.
 func (f *fetchResult) SetReceiptsDone() {
-	if v := atomic.LoadInt32(&f.pending); v == 2 || v == 3 {
+	if v := atomic.LoadInt32(&f.pending); (v & 2) != 0 {
 		atomic.AddInt32(&f.pending, -2)
 	}
 }
 
-// CheckDone checks if the given type is done already
+// Done 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)
+		return (v & 1) == 0
 	case ReceiptType:
-		return !(v == 2 || v == 3)
+		return (v & 2) == 0
 	default:
 		return false
 	}
@@ -148,23 +148,19 @@ type queue struct {
 // newQueue creates a new download queue for scheduling block retrieval.
 func newQueue(blockCacheLimit int) *queue {
 	lock := new(sync.RWMutex)
-	return &queue{
+	q := &queue{
-		headerPendPool:   make(map[string]*fetchRequest),
 		headerContCh:     make(chan bool),
-		blockTaskPool:    make(map[common.Hash]*types.Header),
 		blockTaskQueue:   prque.New(nil),
-		blockPendPool:    make(map[string]*fetchRequest),
-		receiptTaskPool:  make(map[common.Hash]*types.Header),
 		receiptTaskQueue: prque.New(nil),
-		receiptPendPool:  make(map[string]*fetchRequest),
-		resultCache:      newResultStore(blockCacheLimit * 2),
 		active:           sync.NewCond(lock),
 		lock:             lock,
 	}
+	q.Reset(blockCacheLimit)
+	return q
 }
 
 // Reset clears out the queue contents.
-func (q *queue) Reset() {
+func (q *queue) Reset(blockCacheLimit int) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
@@ -182,7 +178,7 @@ func (q *queue) Reset() {
 	q.receiptTaskQueue.Reset()
 	q.receiptPendPool = make(map[string]*fetchRequest)
 
-	q.resultCache = newResultStore(blockCacheItems * 2)
+	q.resultCache = newResultStore(blockCacheLimit * 2)
 }
 
 // Close marks the end of the sync, unblocking Results.
@@ -190,8 +186,8 @@ func (q *queue) Reset() {
 func (q *queue) Close() {
 	q.lock.Lock()
 	q.closed = true
+	q.active.Signal()
 	q.lock.Unlock()
-	q.active.Broadcast()
 }
 
 // PendingHeaders retrieves the number of header requests pending for retrieval.
@@ -341,20 +337,15 @@ 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
+// Results can be called concurrently with Deliver and Schedule,
-// callers to Results (both will modify q.resultSize)
+// but assumes that there are not two simultaneous callers to Results
 func (q *queue) Results(block bool) []*fetchResult {
-
 	// abort early if there are no items and non-blocking requested
-	if !q.resultCache.HasCompletedItems() && !block {
+	if !block && !q.resultCache.HasCompletedItems() {
-		return nil
-	}
-	results := q.resultCache.GetCompleted(maxResultsProcess)
-	if len(results) == 0 && !block {
 		return nil
 	}
 	closed := false
-	for !closed && len(results) == 0 {
+	for !closed && !q.resultCache.HasCompletedItems() {
 		// In order to wait on 'active', we need to obtain the lock.
 		// That may take a while, if someone is delivering at the same
 		// time, so after obtaining the lock, we check again if there
@@ -363,17 +354,17 @@ func (q *queue) Results(block bool) []*fetchResult {
 		// someone can have closed the queue. In that case, we should
 		// return the available results and stop blocking
 		q.lock.Lock()
-		closed = q.closed
+		if q.resultCache.HasCompletedItems() || q.closed {
-		results = q.resultCache.GetCompleted(maxResultsProcess)
-		if closed || len(results) > 0 {
 			q.lock.Unlock()
 			break
 		}
+		// No items available, and not closed
 		q.active.Wait()
 		closed = q.closed
 		q.lock.Unlock()
-		results = q.resultCache.GetCompleted(maxResultsProcess)
 	}
+	// Regardless if closed or not, we can still deliver whatever we have
+	results := q.resultCache.GetCompleted(maxResultsProcess)
 	for _, result := range results {
 		// Recalculate the result item weights to prevent memory exhaustion
 		size := result.Header.Size()
@@ -397,7 +388,7 @@ func (q *queue) Results(block bool) []*fetchResult {
 	if time.Now().Second()&0xa == 0 {
 		info := q.Stats()
 		info = append(info, "throttle", throttleThreshold)
-		log.Info("queue stats", info...)
+		log.Info("Downloader queue stats", info...)
 	}
 	return results
 }
@@ -866,6 +857,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 		hashes = append(hashes, header.Hash())
 		i++
 	}
+
 	q.lock.Lock()
 	var acceptCount = 0
 	for _, header := range request.Headers[:i] {
@@ -889,12 +881,11 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 			taskQueue.Push(header, -int64(header.Number.Uint64()))
 		}
 	}
-	q.lock.Unlock()
-
 	// Wake up Results
 	if acceptCount > 0 {
-		q.active.Broadcast()
+		q.active.Signal()
 	}
+	q.lock.Unlock()
 	// If none of the data was good, it's a stale delivery
 	switch {
 	case failure == nil || failure == errInvalidChain:

eth/downloader/queue_test.go

@@ -242,11 +242,11 @@ func TestEmptyBlocks(t *testing.T) {
 
 }
 
-// xTestDelivery does some more extensive testing of events that happen,
+// XTestDelivery does some more extensive testing of events that happen,
 // blocks that become known and peers that make reservations and deliveries.
 // disabled since it's not really a unit-test, but can be executed to test
 // some more advanced scenarios
-func xTestDelivery(t *testing.T) {
+func XTestDelivery(t *testing.T) {
 	// the outside network, holding blocks
 	blo, rec := makeChain(128, 0, genesis, false)
 	world := newNetwork()

eth/downloader/resultcache.go

@@ -25,7 +25,6 @@ import (
 	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/log"
 )
 
 type resultStore struct {
@@ -73,31 +72,10 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) {
 //                prio right now
 // fetchResult -- the result to store data into
 // err         -- any error that occurred
-func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (bool, bool, *fetchResult, error) {
+func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) {
-	r.lock.RLock()
-	var index int
-	item, index, stale, throttled, err := r.getFetchResult(header.Number.Uint64())
-	if err != nil || stale || throttled {
-		r.lock.RUnlock()
-		// Index is above the current threshold of 'prioritized' blocks,
-		if throttled {
-			log.Debug("resultcache throttle", "index", index, "threshold", r.throttleThreshold)
-
-		}
-		return stale, throttled, item, err
-	}
-	if item != nil {
-		// All good, item already exists (perhaps a receipt fetch following
-		// a body fetch)
-		r.lock.RUnlock()
-		return stale, throttled, item, err
-	}
-	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()
-	// Same checks as above, now with wlock
+	var index int
 	item, index, stale, throttled, err = r.getFetchResult(header.Number.Uint64())
 	if err != nil || stale || throttled {
 		return stale, throttled, item, err
@@ -132,13 +110,13 @@ func (r *resultStore) getFetchResult(headerNumber uint64) (item *fetchResult, in
 		err = fmt.Errorf("index allocation went beyond available resultStore space "+
 			"(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d",
 			index, headerNumber, r.resultOffset, len(r.items))
-		return
+		return nil, index, stale, throttle, err
 	}
 	if stale {
-		return
+		return nil, index, stale, throttle, nil
 	}
 	item = r.items[index]
-	return
+	return item, index, stale, throttle, nil
 }
 
 // hasCompletedItems returns true if there are processable items available