downloader: free up peers from work when the downloader resets

eth/downloader/downloader.go

@@ -297,8 +297,9 @@ out:
 
 				// make sure that we have peers available for fetching. If all peers have been tried
 				// and all failed throw an error
-				if len(availablePeers) > 0 && d.queue.fetchPool.Size() == 0 {
+				if len(d.queue.fetching) == 0 {
 					d.queue.reset()
+					d.peers.reset()
 
 					return errPeersUnavailable
 				}
@@ -337,6 +338,7 @@ out:
 					d.queue.deliver(pid, nil)
 					if peer := d.peers[pid]; peer != nil {
 						peer.demote()
+						peer.reset()
 					}
 				}
 

eth/downloader/peer.go

@@ -20,6 +20,12 @@ type blockFetcherFn func([]common.Hash) error
 // XXX make threadsafe!!!!
 type peers map[string]*peer
 
+func (p peers) reset() {
+	for _, peer := range p {
+		p.reset()
+	}
+}
+
 func (p peers) get(state int) []*peer {
 	var peers []*peer
 	for _, peer := range p {
@@ -128,3 +134,7 @@ func (p *peer) demote() {
 		p.rep = 0
 	}
 }
+
+func (p *peer) reset() {
+	p.state = idleState
+}

eth/downloader/queue.go

@@ -60,13 +60,18 @@ func (c *queue) get(p *peer, max int) *chunk {
 			return false
 		}
 
-		hashes.Add(v)
-		i++
+		// Skip any hashes that have previously been requested from the peer
+		if !p.requested.Has(v) {
+			hashes.Add(v)
+			i++
+		}
 
 		return true
 	})
-	// remove hashes that have previously been fetched
-	hashes.Separate(p.requested)
+	// if no hashes can be requested return a nil chunk
+	if hashes.Size() == 0 {
+		return nil
+	}
 
 	// remove the fetchable hashes from hash pool
 	c.hashPool.Separate(hashes)