eth/downloader: add back some locks, address review concerns

core/types/block.go

@@ -137,12 +137,12 @@ func rlpHash(x interface{}) (h common.Hash) {
 }
 
 // EmptyBody returns true if there is no additional 'body' to complete the header
-// that is: no transactions and no uncles
+// 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
+// EmptyReceipts returns true if there are no receipts for this header/block.
 func (h *Header) EmptyReceipts() bool {
 	return h.ReceiptHash == EmptyRootHash
 }

eth/downloader/downloader.go

@@ -511,9 +511,9 @@ func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.I
 		d.syncInitHook(origin, height)
 	}
 	fetchers := []func() error{
-		func() error { return d.fetchHeaders(p, origin+1, pivot, height) }, // Headers are always retrieved
+		func() error { return d.fetchHeaders(p, origin+1, pivot) }, // Headers are always retrieved
-		func() error { return d.fetchBodies(origin + 1) },                  // Bodies are retrieved during normal and fast sync
+		func() error { return d.fetchBodies(origin + 1) },          // Bodies are retrieved during normal and fast sync
-		func() error { return d.fetchReceipts(origin + 1) },                // Receipts are retrieved during fast sync
+		func() error { return d.fetchReceipts(origin + 1) },        // Receipts are retrieved during fast sync
 		func() error { return d.processHeaders(origin+1, pivot, td) },
 	}
 	if d.mode == FastSync {
@@ -621,7 +621,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.Info("Multiple headers for single request", "headers", len(headers))
+				p.log.Warn("Multiple headers for single request", "headers", len(headers))
 				return nil, errBadPeer
 			}
 			head := headers[0]
@@ -853,7 +853,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.Info("Multiple headers for single request", "headers", len(headers))
+					p.log.Warn("Multiple headers for single request", "headers", len(headers))
 					return 0, errBadPeer
 				}
 				arrived = true
@@ -877,7 +877,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.Info("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check)
+					p.log.Warn("Received non requested header", "number", header.Number, "hash", header.Hash(), "request", check)
 					return 0, errBadPeer
 				}
 				start = check
@@ -910,7 +910,7 @@ func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header)
 // other peers are only accepted if they map cleanly to the skeleton. If no one
 // can fill in the skeleton - not even the origin peer - it's assumed invalid and
 // the origin is dropped.
-func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead uint64) error {
+func (d *Downloader) fetchHeaders(p *peerConnection, from uint64, pivot uint64) error {
 	p.log.Debug("Directing header downloads", "origin", from)
 	defer p.log.Debug("Header download terminated")
 
@@ -927,6 +927,7 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 
 		ttl = d.requestTTL()
 		timeout.Reset(ttl)
+
 		if skeleton {
 			p.log.Trace("Fetching skeleton headers", "count", MaxHeaderFetch, "from", from)
 			go p.peer.RequestHeadersByNumber(from+uint64(MaxHeaderFetch)-1, MaxSkeletonSize, MaxHeaderFetch-1, false)
@@ -982,7 +983,7 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 				}
 			}
 			headers := packet.(*headerPack).headers
-			ignoredHeaders := 0
+
 			// If we received a skeleton batch, resolve internals concurrently
 			if skeleton {
 				filled, proced, err := d.fillHeaderSkeleton(from, headers)
@@ -1020,7 +1021,6 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 							delay = n
 						}
 						headers = headers[:n-delay]
-						ignoredHeaders = delay
 					}
 				}
 			}
@@ -1035,15 +1035,8 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 				from += uint64(len(headers))
 				getHeaders(from)
 			} else {
-				// No headers delivered
-				if !skeleton && ignoredHeaders == 0 && from < advertisedHead {
-					// This peer told is about a high number, but is not
-					// delivering
-					p.log.Trace("peer did not deliver", "requested", from, "head", advertisedHead)
-					return errStallingPeer
-				}
-				p.log.Trace("All headers delayed, waiting")
 				// No headers delivered, or all of them being delayed, sleep a bit and retry
+				p.log.Trace("All headers delayed, waiting")
 				select {
 				case <-time.After(fsHeaderContCheck):
 					getHeaders(from)
@@ -1328,7 +1321,7 @@ func (d *Downloader) fetchParts(deliveryCh chan dataPack, deliver func(dataPack)
 				// have them.
 				request, progress, throttle, err := reserve(peer, capacity(peer))
 				if err != nil {
-					log.Info("Error in loop", "err", err)
+					log.Info("Error reserving fetch", "err", err)
 					return err
 				}
 				if progress {
@@ -1489,7 +1482,8 @@ 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, "parent", chunk[n].ParentHash)
+						log.Info("Invalid header encountered", "number", chunk[n].Number,
+							"hash", chunk[n].Hash(), "parent", chunk[n].ParentHash, "err", err)
 						return errInvalidChain
 					}
 					// All verifications passed, store newly found uncertain headers
@@ -1682,12 +1676,11 @@ func (d *Downloader) processFastSyncContent(latest *types.Header) error {
 
 func splitAroundPivot(pivot uint64, results []*fetchResult) (p *fetchResult, before, after []*fetchResult) {
 	if len(results) == 0 {
-		return
+		return nil, nil, nil
 	}
 	if lastNum := results[len(results)-1].Header.Number.Uint64(); lastNum < pivot {
 		// the pivot is somewhere in the future
-		before = results
+		return nil, results, nil
-		return
 	}
 	// This can also be optimized, but only happens very seldom
 	for _, result := range results {

eth/downloader/peer.go

@@ -115,15 +115,11 @@ func (w *lightPeerWrapper) RequestNodeData([]common.Hash) error {
 // newPeerConnection creates a new downloader peer.
 func newPeerConnection(id string, version int, peer Peer, logger log.Logger) *peerConnection {
 	return &peerConnection{
-		id:                id,
+		id:      id,
-		lacking:           make(map[common.Hash]struct{}),
+		lacking: make(map[common.Hash]struct{}),
-		peer:              peer,
+		peer:    peer,
-		version:           version,
+		version: version,
-		log:               logger,
+		log:     logger,
-		headerThroughput:  float64(MaxHeaderFetch / 8),
-		blockThroughput:   float64(MaxBlockFetch / 8),
-		receiptThroughput: float64(MaxReceiptFetch / 8),
-		stateThroughput:   float64(MaxStateFetch / 8),
 	}
 }
 
@@ -137,10 +133,10 @@ func (p *peerConnection) Reset() {
 	atomic.StoreInt32(&p.receiptIdle, 0)
 	atomic.StoreInt32(&p.stateIdle, 0)
 
-	p.headerThroughput = float64(MaxHeaderFetch / 8)
+	p.headerThroughput = 0
-	p.blockThroughput = float64(MaxBlockFetch / 8)
+	p.blockThroughput = 0
-	p.receiptThroughput = float64(MaxReceiptFetch / 8)
+	p.receiptThroughput = 0
-	p.stateThroughput = float64(MaxStateFetch / 8)
+	p.stateThroughput = 0
 
 	p.lacking = make(map[common.Hash]struct{})
 }

eth/downloader/peer_test.go

@@ -1,3 +1,19 @@
+// Copyright 2020 The go-ethereum Authors
+// This file is part of go-ethereum.
+//
+// go-ethereum is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// go-ethereum 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.
+
 package downloader
 
 import (

eth/downloader/queue.go

@@ -34,8 +34,8 @@ import (
 )
 
 const (
-	BodyType    = 0
+	bodyType    = uint(0)
-	ReceiptType = 1
+	receiptType = uint(1)
 )
 
 var (
@@ -73,44 +73,37 @@ func newFetchResult(header *types.Header, fastSync bool) *fetchResult {
 		Header: header,
 	}
 	if !header.EmptyBody() {
-		item.pending = 1
+		item.pending |= (1 << bodyType)
 	}
 	if fastSync && !header.EmptyReceipts() {
-		item.pending += 2
+		item.pending |= (1 << receiptType)
 	}
 	return item
 }
 
 // SetBodyDone flags the body as finished.
 func (f *fetchResult) SetBodyDone() {
-	if v := atomic.LoadInt32(&f.pending); (v & 1) != 0 {
+	if v := atomic.LoadInt32(&f.pending); (v & (1 << bodyType)) != 0 {
 		atomic.AddInt32(&f.pending, -1)
 	}
 }
 
-// AllDone checks item is done
+// AllDone checks if 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) != 0 {
+	if v := atomic.LoadInt32(&f.pending); (v & (1 << receiptType)) != 0 {
 		atomic.AddInt32(&f.pending, -2)
 	}
 }
 
 // Done checks if the given type is done already
-func (f *fetchResult) Done(typ int) bool {
+func (f *fetchResult) Done(kind uint) bool {
 	v := atomic.LoadInt32(&f.pending)
-	switch typ {
+	return v&(1<<kind) == 0
-	case BodyType:
-		return (v & 1) == 0
-	case ReceiptType:
-		return (v & 2) == 0
-	default:
-		return false
-	}
 }
 
 // queue represents hashes that are either need fetching or are being fetched
@@ -453,7 +446,7 @@ func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, BodyType)
+	return q.reserveHeaders(p, count, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, bodyType)
 }
 
 // ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
@@ -463,7 +456,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, ReceiptType)
+	return q.reserveHeaders(p, count, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, receiptType)
 }
 
 // reserveHeaders reserves a set of data download operations for a given peer,
@@ -479,7 +472,7 @@ func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bo
 // 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, typ int) (*fetchRequest, bool, bool, error) {
+	pendPool map[string]*fetchRequest, kind uint) (*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() {
@@ -496,7 +489,8 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 	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)
+		h, _ := taskQueue.Peek()
+		header := h.(*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
 
@@ -504,14 +498,15 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
 		if stale {
 			// Don't put back in the task queue, this item has already been
 			// delivered upstream
+			taskQueue.PopItem()
 			progress = true
 			delete(taskPool, header.Hash())
 			proc = proc - 1
+			log.Error("Fetch reservation already delivered", "number", header.Number.Uint64())
 			continue
 		}
 		if throttle {
-			// There are no resultslots available. Put it back in the task queue
+			// There are no resultslots available. Leave it in the task queue
-			taskQueue.Push(header, -int64(header.Number.Uint64()))
 			// 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
@@ -521,17 +516,19 @@ 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)
-			// There are no resultslots available. Put it back in the task queue
+			// There are no resultslots available. Leave it in the task queue
-			taskQueue.Push(header, -int64(header.Number.Uint64()))
 			break
 		}
-		if item.Done(typ) {
+		if item.Done(kind) {
 			// If it's a noop, we can skip this task
 			delete(taskPool, header.Hash())
+			taskQueue.PopItem()
 			proc = proc - 1
 			progress = true
 			continue
 		}
+		// Remove it from the task queue
+		taskQueue.PopItem()
 		// Otherwise unless the peer is known not to have the data, add to the retrieve list
 		if p.Lacks(header.Hash()) {
 			skip = append(skip, header)
@@ -673,22 +670,6 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest,
 	return expiries
 }
 
-func (q *queue) getHeaderRequest(id string) (request *fetchRequest, target common.Hash, err error) {
-	q.lock.Lock()
-	defer q.lock.Unlock()
-	// Short circuit if the data was never requested
-	request = q.headerPendPool[id]
-	if request == nil {
-		return nil, target, errNoFetchesPending
-	}
-	headerReqTimer.UpdateSince(request.Time)
-	delete(q.headerPendPool, id)
-	// Ensure headers can be mapped onto the skeleton chain
-	target = q.headerTaskPool[request.From].Hash()
-	return request, target, nil
-
-}
-
 // DeliverHeaders injects a header retrieval response into the header results
 // cache. This method either accepts all headers it received, or none of them
 // if they do not map correctly to the skeleton.
@@ -697,10 +678,20 @@ func (q *queue) getHeaderRequest(id string) (request *fetchRequest, target commo
 // of ready headers to the processor to keep the pipeline full. However it will
 // not block to prevent stalling other pending deliveries.
 func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh chan []*types.Header) (int, error) {
-	request, target, err := q.getHeaderRequest(id)
+	q.lock.Lock()
-	if err != nil {
+	defer q.lock.Unlock()
-		return 0, err
+
+	// Short circuit if the data was never requested
+	request := q.headerPendPool[id]
+	if request == nil {
+		return 0, errNoFetchesPending
 	}
+	headerReqTimer.UpdateSince(request.Time)
+	delete(q.headerPendPool, id)
+
+	// Ensure headers can be mapped onto the skeleton chain
+	target := q.headerTaskPool[request.From].Hash()
+
 	accepted := len(headers) == MaxHeaderFetch
 	if accepted {
 		if headers[0].Number.Uint64() != request.From {
@@ -726,8 +717,6 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 			}
 		}
 	}
-	q.lock.Lock()
-	defer q.lock.Unlock()
 	// If the batch of headers wasn't accepted, mark as unavailable
 	if !accepted {
 		log.Trace("Skeleton filling not accepted", "peer", id, "from", request.From)
@@ -773,7 +762,8 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
 // The method returns the number of blocks bodies accepted from the delivery and
 // also wakes any threads waiting for data delivery.
 func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) {
-
+	q.lock.Lock()
+	defer q.lock.Unlock()
 	validate := func(index int, header *types.Header) error {
 		if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash {
 			return errInvalidBody
@@ -797,7 +787,8 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi
 // The method returns the number of transaction receipts accepted from the delivery
 // and also wakes any threads waiting for data delivery.
 func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, error) {
-
+	q.lock.Lock()
+	defer q.lock.Unlock()
 	validate := func(index int, header *types.Header) error {
 		if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash {
 			return errInvalidReceipt
@@ -814,23 +805,21 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int,
 
 // deliver injects a data retrieval response into the results queue.
 //
-// This method obtains the lock as needed
+// Note, this method expects the queue lock to be already held for writing. The
+// reason this 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) deliver(id string, taskPool map[common.Hash]*types.Header,
 	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) {
 
-	q.lock.Lock()
 	// Short circuit if the data was never requested
 	request := pendPool[id]
 	if request == nil {
-		q.lock.Unlock()
 		return 0, errNoFetchesPending
 	}
 	reqTimer.UpdateSince(request.Time)
 	delete(pendPool, id)
-	// Now we have exclusive access to 'request', and can unlock
-	q.lock.Unlock()
 
 	// If no data items were retrieved, mark them as unavailable for the origin peer
 	if results == 0 {
@@ -867,19 +856,16 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 			// else: betweeen here and above, some other peer filled this result,
 			// or it was indeed a no-op. This should not happen, but if it does it's
 			// not something to panic about
-			log.Info("delivery stale?", "err", err, "stale", stale)
+			log.Error("Delivery stale", "stale", stale, "number", header.Number.Uint64(), "err", err)
 			failure = errStaleDelivery
 		}
 		// Clean up a successful fetch
 		delete(taskPool, hashes[acceptCount])
-		request.Headers[acceptCount] = nil
 		acceptCount++
 	}
 	// Return all failed or missing fetches to the queue
 	for _, header := range request.Headers[acceptCount:] {
-		if header != nil {
+		taskQueue.Push(header, -int64(header.Number.Uint64()))
-			taskQueue.Push(header, -int64(header.Number.Uint64()))
-		}
 	}
 	// Wake up Results
 	if acceptCount > 0 {