eth/downloader/fetchers_concurrent.go

Type	MOD
Upstream Lines	369
Changed	+28 -4

Add header fetch type to concurrent fetcher framework

diff --git a/eth/downloader/fetchers_concurrent.go b/eth/downloader/fetchers_concurrent.go
index 51bf3404b..2bf761c73 100644
--- a/eth/downloader/fetchers_concurrent.go
+++ b/eth/downloader/fetchers_concurrent.go
@@ -126,6 +126,10 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
    // Prepare the queue and fetch block parts until the block header fetcher's done
    finished := false
    for {
+       // Short circuit if we lost all our peers
+       if d.peers.Len() == 0 {
+           return errNoPeers
+       }
        // If there's nothing more to fetch, wait or terminate
        if queue.pending() == 0 {
            if len(pending) == 0 && finished {
@@ -134,8 +138,10 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
        } else {
            // Send a download request to all idle peers, until throttled
            var (
-               idles []*peerConnection
-               caps  []int
+               idles      []*peerConnection
+               caps       []int
+               progressed bool
+               queued     = queue.pending()
            )
            for _, peer := range d.peers.AllPeers() {
                pending, stale := pending[peer.id], stales[peer.id]
@@ -161,13 +167,16 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
                if throttled {
                    break
                }
-               if queued := queue.pending(); queued == 0 {
+               if queued = queue.pending(); queued == 0 {
                    break
                }
                // Reserve a chunk of fetches for a peer. A nil can mean either that
                // no more headers are available, or that the peer is known not to
                // have them.
-               request, _, throttle := queue.reserve(peer, queue.capacity(peer, d.peers.rates.TargetRoundTrip()))
+               request, progress, throttle := queue.reserve(peer, queue.capacity(peer, d.peers.rates.TargetRoundTrip()))
+               if progress {
+                   progressed = true
+               }
                if throttle {
                    throttled = true
                    throttleCounter.Inc(1)
@@ -196,6 +205,11 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
                    timeout.Reset(ttl)
                }
            }
+           // Make sure that we have peers available for fetching. If all peers have been tried
+           // and all failed throw an error
+           if !progressed && !throttled && len(pending) == 0 && len(idles) == d.peers.Len() && queued > 0 {
+               return errPeersUnavailable
+           }
        }
        // Wait for something to happen
        select {
@@ -299,6 +313,16 @@ func (d *Downloader) concurrentFetch(queue typedQueue) error {
                queue.updateCapacity(peer, 0, 0)
            } else {
                d.dropPeer(peer.id)
+
+               // If this peer was the master peer, abort sync immediately
+               d.cancelLock.RLock()
+               master := peer.id == d.cancelPeer
+               d.cancelLock.RUnlock()
+
+               if master {
+                   d.cancel()
+                   return errTimeout
+               }
            }

        case res := <-responses:

← Back to Sync & Downloader