forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 09:23:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge aa11910842 into 5d4267911a

Browse source
This commit is contained in:
Felix Lange 2018-01-22 22:50:53 +00:00 committed by GitHub
commit 4e0ee2a9e9
1 changed files with 77 additions and 99 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

176
eth/downloader/downloader.go
View file

@ -993,15 +993,12 @@ func (d *Downloader) fetchParts(errCancel error, deliveryCh chan dataPack, deliv
ticker := time.NewTicker(100 * time.Millisecond) ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop() defer ticker.Stop()
update := make(chan struct{}, 1)
// Prepare the queue and fetch block parts until the block header fetcher's done // Prepare the queue and fetch block parts until the block header fetcher's done
finished := false finished := false
for { for {
select { select {
case <-d.cancelCh: case <-d.cancelCh:
return errCancel return errCancel
case packet := <-deliveryCh: case packet := <-deliveryCh:
// If the peer was previously banned and failed to deliver its pack // If the peer was previously banned and failed to deliver its pack
// in a reasonable time frame, ignore its message. // in a reasonable time frame, ignore its message.
@ -1027,115 +1024,96 @@ func (d *Downloader) fetchParts(errCancel error, deliveryCh chan dataPack, deliv
peer.log.Trace("Failed to deliver retrieved data", "type", kind, "err", err) peer.log.Trace("Failed to deliver retrieved data", "type", kind, "err", err)
} }
} }
// Blocks assembled, try to update the progress
select {
case update <- struct{}{}:
default:
}
case cont := <-wakeCh: case cont := <-wakeCh:
// The header fetcher sent a continuation flag, check if it's done // The header fetcher sent a continuation flag, check if it's done
if !cont { if !cont {
finished = true finished = true
} }
// Headers arrive, try to update the progress
select {
case update <- struct{}{}:
default:
}
case <-ticker.C: case <-ticker.C:
// Sanity check update the progress }
select {
case update <- struct{}{}:
default:
}
case <-update: // Short circuit if we lost all our peers
// Short circuit if we lost all our peers if d.peers.Len() == 0 {
if d.peers.Len() == 0 { return errNoPeers
return errNoPeers }
} // Check for fetch request timeouts and demote the responsible peers
// Check for fetch request timeouts and demote the responsible peers for pid, fails := range expire() {
for pid, fails := range expire() { if peer := d.peers.Peer(pid); peer != nil {
if peer := d.peers.Peer(pid); peer != nil { // If a lot of retrieval elements expired, we might have overestimated the remote peer or perhaps
// If a lot of retrieval elements expired, we might have overestimated the remote peer or perhaps // ourselves. Only reset to minimal throughput but don't drop just yet. If even the minimal times
// ourselves. Only reset to minimal throughput but don't drop just yet. If even the minimal times // out that sync wise we need to get rid of the peer.
// out that sync wise we need to get rid of the peer. //
// // The reason the minimum threshold is 2 is because the downloader tries to estimate the bandwidth
// The reason the minimum threshold is 2 is because the downloader tries to estimate the bandwidth // and latency of a peer separately, which requires pushing the measures capacity a bit and seeing
// and latency of a peer separately, which requires pushing the measures capacity a bit and seeing // how response times reacts, to it always requests one more than the minimum (i.e. min 2).
// how response times reacts, to it always requests one more than the minimum (i.e. min 2). if fails > 2 {
if fails > 2 { peer.log.Trace("Data delivery timed out", "type", kind)
peer.log.Trace("Data delivery timed out", "type", kind) setIdle(peer, 0)
setIdle(peer, 0) } else {
} else { peer.log.Debug("Stalling delivery, dropping", "type", kind)
peer.log.Debug("Stalling delivery, dropping", "type", kind) d.dropPeer(pid)
d.dropPeer(pid)
}
} }
} }
// If there's nothing more to fetch, wait or terminate }
if pending() == 0 { // If there's nothing more to fetch, wait or terminate
if !inFlight() && finished { if pending() == 0 {
log.Debug("Data fetching completed", "type", kind) if !inFlight() && finished {
return nil log.Debug("Data fetching completed", "type", kind)
} return nil
}
continue
}
// Send a download request to all idle peers, until throttled
progressed, throttled, running := false, false, inFlight()
idles, total := idle()
for _, peer := range idles {
// Short circuit if throttling activated
if throttle() {
throttled = true
break break
} }
// Send a download request to all idle peers, until throttled // Short circuit if there is no more available task.
progressed, throttled, running := false, false, inFlight() if pending() == 0 {
idles, total := idle() break
for _, peer := range idles {
// Short circuit if throttling activated
if throttle() {
throttled = true
break
}
// Short circuit if there is no more available task.
if pending() == 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, progress, err := reserve(peer, capacity(peer))
if err != nil {
return err
}
if progress {
progressed = true
}
if request == nil {
continue
}
if request.From > 0 {
peer.log.Trace("Requesting new batch of data", "type", kind, "from", request.From)
} else if len(request.Headers) > 0 {
peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Headers), "from", request.Headers[0].Number)
} else {
peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Hashes))
}
// Fetch the chunk and make sure any errors return the hashes to the queue
if fetchHook != nil {
fetchHook(request.Headers)
}
if err := fetch(peer, request); err != nil {
// Although we could try and make an attempt to fix this, this error really
// means that we've double allocated a fetch task to a peer. If that is the
// case, the internal state of the downloader and the queue is very wrong so
// better hard crash and note the error instead of silently accumulating into
// a much bigger issue.
panic(fmt.Sprintf("%v: %s fetch assignment failed", peer, kind))
}
running = true
} }
// Make sure that we have peers available for fetching. If all peers have been tried // Reserve a chunk of fetches for a peer. A nil can mean either that
// and all failed throw an error // no more headers are available, or that the peer is known not to
if !progressed && !throttled && !running && len(idles) == total && pending() > 0 { // have them.
return errPeersUnavailable request, progress, err := reserve(peer, capacity(peer))
if err != nil {
return err
} }
if progress {
progressed = true
}
if request == nil {
continue
}
if request.From > 0 {
peer.log.Trace("Requesting new batch of data", "type", kind, "from", request.From)
} else if len(request.Headers) > 0 {
peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Headers), "from", request.Headers[0].Number)
} else {
peer.log.Trace("Requesting new batch of data", "type", kind, "count", len(request.Hashes))
}
// Fetch the chunk and make sure any errors return the hashes to the queue
if fetchHook != nil {
fetchHook(request.Headers)
}
if err := fetch(peer, request); err != nil {
// Although we could try and make an attempt to fix this, this error really
// means that we've double allocated a fetch task to a peer. If that is the
// case, the internal state of the downloader and the queue is very wrong so
// better hard crash and note the error instead of silently accumulating into
// a much bigger issue.
panic(fmt.Sprintf("%v: %s fetch assignment failed", peer, kind))
}
running = true
}
// Make sure that we have peers available for fetching. If all peers have been tried
// and all failed throw an error
if !progressed && !throttled && !running && len(idles) == total && pending() > 0 {
return errPeersUnavailable
} }
} }
} }