eth/downloader: improved locking

eth/downloader/queue.go

@@ -93,14 +93,14 @@ type queue struct {
 	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.Mutex
+	lock   *sync.RWMutex
 	active *sync.Cond
 	closed bool
 }
 
 // newQueue creates a new download queue for scheduling block retrieval.
 func newQueue() *queue {
-	lock := new(sync.Mutex)
+	lock := new(sync.RWMutex)
 	return &queue{
 		headerPendPool:   make(map[string]*fetchRequest),
 		headerContCh:     make(chan bool),
@@ -675,6 +675,22 @@ 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.
@@ -683,20 +699,10 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest,
 // 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) {
-	q.lock.Lock()
-	defer q.lock.Unlock()
-
-	// Short circuit if the data was never requested
-	request := q.headerPendPool[id]
-	if request == nil {
-		return 0, errNoFetchesPending
+	request, target, err := q.getHeaderRequest(id)
+	if err != nil {
+		return 0, err
 	}
-	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 {
@@ -722,6 +728,8 @@ 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)
@@ -767,53 +775,59 @@ 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()
 
-	reconstruct := func(header *types.Header, index int, result *fetchResult) error {
-		if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash || types.CalcUncleHash(uncleLists[index]) != header.UncleHash {
+	validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
+		if types.DeriveSha(types.Transactions(txLists[index])) != txHash {
+			return errInvalidBody
+		}
+		if types.CalcUncleHash(uncleLists[index]) != uncleHash {
 			return errInvalidBody
 		}
-		result.Transactions = txLists[index]
-		result.Uncles = uncleLists[index]
 		return nil
 	}
-	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), reconstruct)
+
+	reconstruct := func(index int, result *fetchResult) {
+		result.Transactions = txLists[index]
+		result.Uncles = uncleLists[index]
+	}
+	return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct)
 }
 
 // DeliverReceipts injects a receipt retrieval response into the results queue.
 // 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()
 
-	reconstruct := func(header *types.Header, index int, result *fetchResult) error {
-		if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash {
+	validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
+		if types.DeriveSha(types.Receipts(receiptList[index])) != receiptHash {
 			return errInvalidReceipt
 		}
-		result.Receipts = receiptList[index]
 		return nil
 	}
-	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), reconstruct)
+	reconstruct := func(index int, result *fetchResult) {
+		result.Receipts = receiptList[index]
+	}
+	return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct)
 }
 
 // deliver injects a data retrieval response into the results queue.
 //
-// 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.
+// 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, reconstruct func(header *types.Header, index int, result *fetchResult) error) (int, error) {
+	results int, validate func(index int, txHash, uncleHash, receiptHash common.Hash) 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 {
@@ -823,54 +837,68 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
 	}
 	// Assemble each of the results with their headers and retrieved data parts
 	var (
-		accepted int
-		failure  error
-		useful   bool
+		failure error
+		i       int
 	)
-	for i, header := range request.Headers {
+	// 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
 		}
-		// Reconstruct the next result if contents match up
-		index := int(header.Number.Int64() - int64(q.resultOffset))
-		if index >= len(q.resultCache) || index < 0 || q.resultCache[index] == nil {
-			failure = errInvalidChain
-			break
-		}
-		if err := reconstruct(header, i, q.resultCache[index]); err != nil {
+		// Validate the fields
+		if err := validate(i, header.TxHash, header.UncleHash, header.ReceiptHash); err != nil {
 			failure = err
 			break
 		}
-		hash := header.Hash()
-
-		donePool[hash] = struct{}{}
-		q.resultCache[index].Pending--
-		useful = true
-		accepted++
+		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))
+		if index >= len(q.resultCache) || index < 0 {
+			// TODO! this should probably be errStaleDelivery instead
+			failure = errStaleDelivery
+			break
+		}
+		if res := q.resultCache[index]; res != nil {
+			hash := header.Hash()
+			donePool[hash] = struct{}{}
+			reconstruct(acceptCount, res)
+			res.Pending--
+			delete(taskPool, hash)
+		}
+		// else: betweeen here and above, some other peer filled this result
+		// we just ignore and move on
 
 		// Clean up a successful fetch
-		request.Headers[i] = nil
-		delete(taskPool, hash)
+		request.Headers[acceptCount] = nil
+		acceptCount++
 	}
 	// Return all failed or missing fetches to the queue
-	for _, header := range request.Headers {
+	for _, header := range request.Headers[acceptCount:] {
 		if header != nil {
 			taskQueue.Push(header, -int64(header.Number.Uint64()))
 		}
 	}
+	q.lock.Unlock()
+
 	// Wake up Results
-	if accepted > 0 {
+	if acceptCount > 0 {
 		q.active.Signal()
 	}
 	// If none of the data was good, it's a stale delivery
 	switch {
 	case failure == nil || failure == errInvalidChain:
-		return accepted, failure
-	case useful:
-		return accepted, fmt.Errorf("partial failure: %v", failure)
+		return acceptCount, failure
+	case acceptCount > 0:
+		return acceptCount, fmt.Errorf("partial failure: %v", failure)
 	default:
-		return accepted, errStaleDelivery
+		return acceptCount, errStaleDelivery
 	}
 }