eth/downloader: nitpick fixes

Author: karalabe

eth/downloader/downloader.go

@@ -601,6 +601,7 @@ func (d *Downloader) Terminate() {
 		d.stateBloom.Close()
 	}
 	d.quitLock.Unlock()
+
 	// Cancel any pending download requests
 	d.Cancel()
 }
@@ -1211,6 +1212,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 	// Create a ticker to detect expired retrieval tasks
 	ticker := time.NewTicker(100 * time.Millisecond)
 	defer ticker.Stop()
+
 	update := make(chan struct{}, 1)
 
 	// Prepare the queue and fetch block parts until the block header fetcher's done

eth/downloader/queue.go

@@ -335,7 +335,7 @@ func (q *queue) Schedule(headers []*types.Header, from uint64) []*types.Header {
 // Results can be called concurrently with Deliver and Schedule,
 // 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
+	// Abort early if there are no items and non-blocking requested
 	if !block && !q.resultCache.HasCompletedItems() {
 		return nil
 	}
@@ -379,7 +379,8 @@ func (q *queue) Results(block bool) []*fetchResult {
 	// on the result cache
 	throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize)
 	q.resultCache.SetThrottleThreshold(throttleThreshold)
-	// log some info at certain times
+
+	// Log some info at certain times
 	if time.Since(q.lastStatLog) > 10*time.Second {
 		q.lastStatLog = time.Now()
 		info := q.Stats()
@@ -392,13 +393,15 @@ func (q *queue) Results(block bool) []*fetchResult {
 func (q *queue) Stats() []interface{} {
 	q.lock.RLock()
 	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,
+		"receiptTasks", q.receiptTaskQueue.Size(),
+		"blockTasks", q.blockTaskQueue.Size(),
+		"itemSize", q.resultSize,
 	}
 }
 
@@ -469,11 +472,11 @@ 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
+//
+// Returns:
+//   item     - the fetchRequest
+//   progress - whether any progress was made
+//   throttle - if the caller should throttle for a while
 func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
 	pendPool map[string]*fetchRequest, kind uint) (*fetchRequest, bool, bool) {
 	// Short circuit if the pool has been depleted, or if the peer's already
@@ -518,7 +521,7 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 		}
 		if err != nil {
 			// this most definitely should _not_ happen
-			log.Warn("reserve headers error", "error", err)
+			log.Warn("Failed to reserve headers", "err", err)
 			// There are no resultslots available. Leave it in the task queue
 			break
 		}

eth/downloader/queue_test.go

@@ -75,13 +75,15 @@ func init() {
 	blocks, _ = makeChain(targetBlocks, 0, genesis, true)
 	emptyChain = &chainData{blocks, 0}
 }
+
 func (chain *chainData) headers() []*types.Header {
 	hdrs := make([]*types.Header, len(chain.blocks))
 	for i, b := range chain.blocks {
 		hdrs[i] = b.Header()
 	}
 	return hdrs
 }
+
 func (chain *chainData) Len() int {
 	return len(chain.blocks)
 }
@@ -95,7 +97,6 @@ func dummyPeer(id string) *peerConnection {
 }
 
 func TestBasics(t *testing.T) {
-
 	q := newQueue(10)
 	if !q.Idle() {
 		t.Errorf("new queue should be idle")
@@ -121,10 +122,7 @@ func TestBasics(t *testing.T) {
 	// queue that a certain peer will deliver them for us
 	{
 		peer := dummyPeer("peer-1")
-		fetchReq, _, throttle, err := q.ReserveBodies(peer, 50)
-		if err != nil {
-			t.Fatal(err)
-		}
+		fetchReq, _, throttle := q.ReserveBodies(peer, 50)
 		if !throttle {
 			// queue size is only 10, so throttling should occur
 			t.Fatal("should throttle")
@@ -139,10 +137,8 @@ func TestBasics(t *testing.T) {
 	}
 	{
 		peer := dummyPeer("peer-2")
-		fetchReq, _, throttle, err := q.ReserveBodies(peer, 50)
-		if err != nil {
-			t.Fatal(err)
-		}
+		fetchReq, _, throttle := q.ReserveBodies(peer, 50)
+
 		// The second peer should hit throttling
 		if !throttle {
 			t.Fatalf("should not throttle")
@@ -158,10 +154,7 @@ func TestBasics(t *testing.T) {
 		// The receipt delivering peer should not be affected
 		// by the throttling of body deliveries
 		peer := dummyPeer("peer-3")
-		fetchReq, _, throttle, err := q.ReserveReceipts(peer, 50)
-		if err != nil {
-			t.Fatal(err)
-		}
+		fetchReq, _, throttle := q.ReserveReceipts(peer, 50)
 		if !throttle {
 			// queue size is only 10, so throttling should occur
 			t.Fatal("should throttle")
@@ -181,7 +174,6 @@ func TestBasics(t *testing.T) {
 }
 
 func TestEmptyBlocks(t *testing.T) {
-
 	q := newQueue(10)
 
 	q.Prepare(1, FastSync)
@@ -208,10 +200,8 @@ func TestEmptyBlocks(t *testing.T) {
 	{
 		// Reserve blocks
 		peer := dummyPeer("peer-1")
-		fetchReq, _, _, err := q.ReserveBodies(peer, 50)
-		if err != nil {
-			t.Fatal(err)
-		}
+		fetchReq, _, _ := q.ReserveBodies(peer, 50)
+
 		// there should be nothing to fetch, blocks are empty
 		if fetchReq != nil {
 			t.Fatal("there should be no body fetch tasks remaining")
@@ -227,10 +217,8 @@ func TestEmptyBlocks(t *testing.T) {
 	//fmt.Printf("receiptTaskQueue len: %d\n", q.receiptTaskQueue.Size())
 	{
 		peer := dummyPeer("peer-3")
-		fetchReq, _, _, err := q.ReserveReceipts(peer, 50)
-		if err != nil {
-			t.Fatal(err)
-		}
+		fetchReq, _, _ := q.ReserveReceipts(peer, 50)
+
 		// there should be nothing to fetch, blocks are empty
 		if fetchReq != nil {
 			t.Fatal("there should be no body fetch tasks remaining")
@@ -296,7 +284,7 @@ func XTestDelivery(t *testing.T) {
 		i := 4
 		for {
 			peer := dummyPeer(fmt.Sprintf("peer-%d", i))
-			f, _, _, _ := q.ReserveBodies(peer, rand.Intn(30))
+			f, _, _ := q.ReserveBodies(peer, rand.Intn(30))
 			if f != nil {
 				var emptyList []*types.Header
 				var txs [][]*types.Transaction
@@ -322,7 +310,7 @@ func XTestDelivery(t *testing.T) {
 		// reserve receiptfetch
 		peer := dummyPeer("peer-3")
 		for {
-			f, _, _, _ := q.ReserveReceipts(peer, rand.Intn(50))
+			f, _, _ := q.ReserveReceipts(peer, rand.Intn(50))
 			if f != nil {
 				var rcs [][]*types.Receipt
 				for _, hdr := range f.Headers {

eth/downloader/resultcache.go renamed to eth/downloader/resultstore.go

@@ -14,9 +14,6 @@
 // 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 (
@@ -27,9 +24,10 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
+// resultStore implements a structure for maintaining fetchResults, tracking their
+// download-progress and delivering (finished) results.
 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
 
 	// Internal index of first non-completed entry, updated atomically when needed.
@@ -43,20 +41,24 @@ type resultStore struct {
 	// 8192 possible places. The queue will, at certain times, recalibrate
 	// this index.
 	throttleThreshold uint64
+
+	lock sync.RWMutex
 }
 
 func newResultStore(size int) *resultStore {
 	return &resultStore{
 		resultOffset:      0,
 		items:             make([]*fetchResult, size),
-		lock:              new(sync.RWMutex),
 		throttleThreshold: 3 * uint64(size) / 4, // 75%
 	}
 }
 
+// SetThrottleThreshold updates the throttling threshold based on the requested
+// limit and the total queue capacity.
 func (r *resultStore) SetThrottleThreshold(threshold uint64) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
+
 	limit := uint64(len(r.items)) * 3 / 4
 	if threshold >= limit {
 		threshold = limit
@@ -66,15 +68,16 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) {
 
 // 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
+//   stale     - if true, this item is already passed, and should not be requested again
+//   throttled - if true, the store is at capacity, this particular header is not prio now
+//   item      - 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) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
+
 	var index int
 	item, index, stale, throttled, err = r.getFetchResult(header.Number.Uint64())
 	if err != nil || stale || throttled {
@@ -88,20 +91,20 @@ func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, thro
 }
 
 // GetDeliverySlot returns the fetchResult for the given header. If the 'stale' flag
-// is true, that means the header has already been delivered 'upstream'.
-// This method does not bubble up the 'throttle' flag, since it's moot at the
-// point in time when the item is downloaded and ready for delivery
+// is true, that means the header has already been delivered 'upstream'. This method
+// does not bubble up the 'throttle' flag, since it's moot at the point in time when
+// the item is downloaded and ready for delivery
 func (r *resultStore) GetDeliverySlot(headerNumber uint64) (*fetchResult, bool, error) {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
+
 	res, _, stale, _, err := r.getFetchResult(headerNumber)
 	return res, stale, err
 }
 
-// getFetchResult returns the fetchResult corresponding to the given item, and the index where
-// the result is stored.
+// 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 = index >= int(r.throttleThreshold)
 	stale = index < 0
@@ -124,6 +127,7 @@ func (r *resultStore) getFetchResult(headerNumber uint64) (item *fetchResult, in
 func (r *resultStore) HasCompletedItems() bool {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
+
 	if len(r.items) == 0 {
 		return false
 	}
@@ -135,7 +139,8 @@ func (r *resultStore) HasCompletedItems() bool {
 
 // countCompleted returns the number of items ready for delivery, stopping at
 // the first non-complete item.
-// It assumes (at least) rlock is held
+//
+// The mthod 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
@@ -170,18 +175,19 @@ func (r *resultStore) GetCompleted(limit int) []*fetchResult {
 	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.
+	// Advance the expected block number of the first cache entry
 	r.resultOffset += uint64(limit)
-	// And subtract the number of items from our index
 	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()
+	defer r.lock.Unlock()
+
 	if r.resultOffset < offset {
 		r.resultOffset = offset
 	}
-	r.lock.Unlock()
 }