eth/downloader: fix flaw with checking closed-status causing hang

eth/downloader/downloader.go

@@ -588,8 +588,10 @@ func (d *Downloader) Terminate() {
 	default:
 		close(d.quitCh)
 	}
+	if d.stateBloom != nil {
+		d.stateBloom.Close()
+	}
 	d.quitLock.Unlock()
-
 	// Cancel any pending download requests
 	d.Cancel()
 }
@@ -980,7 +982,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)
@@ -1018,6 +1020,7 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 							delay = n
 						}
 						headers = headers[:n-delay]
+						ignoredHeaders = delay
 					}
 				}
 			}
@@ -1032,7 +1035,8 @@ func (d *Downloader) fetchHeaders(p *peerConnection, from, pivot, advertisedHead
 				from += uint64(len(headers))
 				getHeaders(from)
 			} else {
-				if !skeleton && from < advertisedHead-uint64(reorgProtHeaderDelay) {
+				// 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)
@@ -1198,7 +1202,6 @@ 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

@@ -352,12 +352,27 @@ func (q *queue) Results(block bool) []*fetchResult {
 		return nil
 	}
 	results := q.resultCache.GetCompleted(maxResultsProcess)
-	for len(results) == 0 && !q.closed {
-		if !block {
-			return nil
-		}
+	if len(results) == 0 && !block {
+		return nil
+	}
+	closed := false
+	for !closed && len(results) == 0 {
+		// 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
+		// are any results to fetch.
+		// Also, in-between we ask for the lock and the lock is obtained,
+		// someone can have closed the queue. In that case, we should
+		// return the available results and stop blocking
 		q.lock.Lock()
+		closed = q.closed
+		results = q.resultCache.GetCompleted(maxResultsProcess)
+		if closed || len(results) > 0 {
+			q.lock.Unlock()
+			break
+		}
 		q.active.Wait()
+		closed = q.closed
 		q.lock.Unlock()
 		results = q.resultCache.GetCompleted(maxResultsProcess)
 	}
@@ -880,7 +895,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
 
 	// Wake up Results
 	if acceptCount > 0 {
-		q.active.Signal()
+		q.active.Broadcast()
 	}
 	// If none of the data was good, it's a stale delivery
 	switch {