diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index b1e23f58f..5e9931f59 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -335,7 +335,7 @@ out: // Deliver the received chunk of blocks, but drop the peer if invalid if err := d.queue.Deliver(blockPack.peerId, blockPack.blocks); err != nil { glog.V(logger.Debug).Infof("Failed delivery for peer %s: %v\n", blockPack.peerId, err) - d.peers.Unregister(blockPack.peerId) + peer.Demote() break } if glog.V(logger.Debug) { @@ -358,7 +358,9 @@ out: // 1) Time for them to respond; // 2) Measure their speed; // 3) Amount and availability. - d.peers.Unregister(pid) + if peer := d.peers.Peer(pid); peer != nil { + peer.Demote() + } } // After removing bad peers make sure we actually have sufficient peer left to keep downloading if d.peers.Peers() == 0 { @@ -372,9 +374,13 @@ out: if d.queue.Throttle() { continue } - // Send a download request to all idle peers + // Send a download request to all idle peers, until throttled idlePeers := d.peers.IdlePeers() for _, peer := range idlePeers { + // Short circuit if throttling activated since above + if d.queue.Throttle() { + break + } // Get a possible chunk. If nil is returned no chunk // could be returned due to no hashes available. request := d.queue.Reserve(peer, maxBlockFetch) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index e2dec5571..1ff2d5149 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -86,10 +86,8 @@ func (p *peer) Demote() { for { // Calculate the new reputation value prev := atomic.LoadInt32(&p.rep) - next := prev - 2 - if next < 0 { - next = 0 - } + next := prev / 2 + // Try to update the old value if atomic.CompareAndSwapInt32(&p.rep, prev, next) { return @@ -177,7 +175,7 @@ func (ps *peerSet) AllPeers() []*peer { } // IdlePeers retrieves a flat list of all the currently idle peers within the -// active peer set. +// active peer set, ordered by their reputation. func (ps *peerSet) IdlePeers() []*peer { ps.lock.RLock() defer ps.lock.RUnlock() @@ -188,5 +186,12 @@ func (ps *peerSet) IdlePeers() []*peer { list = append(list, p) } } + for i := 0; i < len(list); i++ { + for j := i + 1; j < len(list); j++ { + if atomic.LoadInt32(&list[i].rep) < atomic.LoadInt32(&list[j].rep) { + list[i], list[j] = list[j], list[i] + } + } + } return list }