diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 84320b65d3..ea18b07e88 100644 --- a/eth/downloader/downloader.go +++ b/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 diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 1bf23914b0..48a0ed6a68 100644 --- a/eth/downloader/queue.go +++ b/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 {